from __future__ import annotations import inspect import uuid import pytest from sqlalchemy import func, select from app.approval.models import ApprovalRequest from app.audit.models import AuditLog from app.audit.service import AuditService from app.core.exceptions import AppException from app.domains.data_access import constants as c from app.domains.data_access.models import ( BINDING_STATUS_ACTIVE, BINDING_STATUS_REVOKED_PENDING, EFFECT_DESIRED_ABSENT, EFFECT_DESIRED_PRESENT, EFFECT_MANAGED_REASON_GRANT_REQUEST, EFFECT_MANAGED_REASON_GROUP_RENAME_BRIDGE, EFFECT_MANAGED_REASON_REVOKE_REQUEST, EFFECT_STATUS_PENDING_RECONCILE, ENTITLEMENT_STATUS_ACTIVE, ENTITLEMENT_VERSION_V1, OWNER_RESOLUTION_FALLBACK, OWNER_RESOLUTION_RESOLVED, OWNER_SOURCE_BOOTSTRAP, RESOURCE_STATUS_ACTIVE, RESOURCE_STATUS_DELETED, DataAccessBinding, DataAccessCompiledEffect, DataAccessEntitlement, DataAccessResource, ) from app.domains.data_access.service import ( DATA_ACCESS_COMPILER_VERSION, DATA_ACCESS_RESOURCE_REMOVAL_REASON_ADMIN_REMOVED, DataAccessRevokeAuthorizationError, DataAccessService, ) from app.domains.email_group.models import EmailGroup, EmailGroupOwner from app.identity.person_models import IdentityPerson from app.integrations.models import Integration async def _data_access_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) async def _add_person(db_session, *, email: str) -> IdentityPerson: row = IdentityPerson( employee_no=f"P{uuid.uuid4().hex[:8]}", work_email=email, display_name=email.split("@", 1)[0], employment_status="active", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:12]}", ) db_session.add(row) await db_session.flush() return row 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, dataset_id: str = "finance", owner_resolution_status: str = OWNER_RESOLUTION_RESOLVED, ) -> DataAccessResource: row = DataAccessResource( integration_id=await _data_access_integration_id(db_session), 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_status, owner_source=OWNER_SOURCE_BOOTSTRAP, source_version="bootstrap-2026-06-10", source_etag=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 = "analytics-readers@example.com", status: str = "active", google_group_id: str | None = "google-group-analytics", ) -> EmailGroup: row = EmailGroup( email=email, name=email.split("@", 1)[0], status=status, google_group_id=google_group_id, ) 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( db_session, *, resource: DataAccessResource, group: EmailGroup, entitlement: DataAccessEntitlement, email_snapshot: str | None = None, ) -> DataAccessBinding: row = 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).lower(), entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, status=BINDING_STATUS_ACTIVE, ) db_session.add(row) await db_session.flush() return row async def _add_effect( db_session, *, binding: DataAccessBinding, resource: DataAccessResource, entitlement: DataAccessEntitlement, desired_state: str = EFFECT_DESIRED_PRESENT, managed_reason: str = EFFECT_MANAGED_REASON_GRANT_REQUEST, target_project_id: str | None = None, target_dataset_id: str | None = None, provider_member: str | None = None, ) -> DataAccessCompiledEffect: row = DataAccessCompiledEffect( binding_id=binding.id, resource_id=binding.resource_id, principal_email_group_id=binding.principal_email_group_id, entitlement_id=binding.entitlement_id, provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, target_project_id=target_project_id or resource.project_id, target_dataset_id=target_dataset_id or resource.dataset_id, provider_member=( provider_member or f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}" f"{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=False, status=EFFECT_STATUS_PENDING_RECONCILE, ) db_session.add(row) await db_session.flush() return row async def _effect_count(db_session) -> int: value = await db_session.scalar( select(func.count()).select_from(DataAccessCompiledEffect), ) return int(value or 0) async def _approval_request_count(db_session) -> int: value = await db_session.scalar( select(func.count()).select_from(ApprovalRequest), ) return int(value or 0) async def test_group_owner_revoke_creates_revoked_pending_absent_effect( db_session, employee_user, ) -> None: group_owner, _ = employee_user resource_owner = await _add_person( db_session, email="resource-owner@example.com", ) entitlement = await _add_entitlement(db_session) resource = await _add_resource( db_session, owner_person_id=resource_owner.id, ) group = await _add_group( db_session, owner_person_id=group_owner.id, ) binding = await _add_binding( db_session, resource=resource, group=group, entitlement=entitlement, ) result = await DataAccessService().revoke_data_access_binding( db_session, actor=group_owner, binding_id=binding.id, reason="group_owner_revoke", trace_id="trace-dax-019-group-owner", ) assert result.revoked_pending_count == 1 assert result.effect_count == 1 assert result.authorized_as == "group_owner" assert binding.status == BINDING_STATUS_REVOKED_PENDING assert binding.revoke_reason == "group_owner_revoke" assert binding.revoked_at is not None effect = await db_session.scalar(select(DataAccessCompiledEffect)) assert effect is not None assert effect.binding_id == binding.id assert effect.desired_state == EFFECT_DESIRED_ABSENT assert effect.provider_member == "group:analytics-readers@example.com" assert effect.provider_role == c.BIGQUERY_DATA_VIEWER_ROLE assert effect.managed_reason == EFFECT_MANAGED_REASON_REVOKE_REQUEST assert effect.status == EFFECT_STATUS_PENDING_RECONCILE assert await _approval_request_count(db_session) == 0 audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.binding.revoke_requested", ), ) assert audit is not None assert audit.trace_id == AuditService._coerce_trace_id( "trace-dax-019-group-owner", ) assert audit.actor_user_id == group_owner.id assert audit.detail is not None assert audit.detail["authorized_as"] == "group_owner" assert audit.detail["effect_count"] == 1 compile_audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.compile.output", AuditLog.resource_id == str(binding.id), ), ) assert compile_audit is not None assert compile_audit.trace_id == audit.trace_id assert compile_audit.detail is not None assert compile_audit.detail["phase"] == "revoke_request" assert compile_audit.detail["effect_count"] == 1 assert compile_audit.detail["authorized_as"] == "group_owner" assert compile_audit.detail["reason"] == "group_owner_revoke" assert compile_audit.detail["effects"] == [ { "effect_id": str(effect.id), "binding_id": str(binding.id), "resource_id": str(resource.id), "principal_email_group_id": str(group.id), "entitlement_id": str(entitlement.id), "provider": c.PROVIDER_BIGQUERY, "resource_type": c.RESOURCE_TYPE_DATASET, "target_project_id": "helix-data-prod", "target_dataset_id": "finance", "provider_member": "group:analytics-readers@example.com", "provider_role": c.BIGQUERY_DATA_VIEWER_ROLE, "desired_state": EFFECT_DESIRED_ABSENT, "entitlement_version_snapshot": ENTITLEMENT_VERSION_V1, "resource_source_version_snapshot": "bootstrap-2026-06-10", "resource_source_etag_snapshot": "etag-finance", "compiler_version": DATA_ACCESS_COMPILER_VERSION, "managed_reason": EFFECT_MANAGED_REASON_REVOKE_REQUEST, "already_present_at_first_apply": False, "status": EFFECT_STATUS_PENDING_RECONCILE, }, ] async def test_resource_owner_and_admins_can_revoke_non_owner_denied( db_session, admin_user, ) -> None: admin, _ = admin_user resource_owner = await _add_person( db_session, email="resource-owner@example.com", ) unrelated = await _add_person(db_session, email="unrelated@example.com") group_owner = await _add_person(db_session, email="group-owner@example.com") entitlement = await _add_entitlement(db_session) resource = await _add_resource( db_session, owner_person_id=resource_owner.id, ) group = await _add_group( db_session, owner_person_id=group_owner.id, ) binding = await _add_binding( db_session, resource=resource, group=group, entitlement=entitlement, ) with pytest.raises(DataAccessRevokeAuthorizationError): await DataAccessService().revoke_data_access_binding( db_session, actor=unrelated, binding_id=binding.id, reason="not_allowed", trace_id="trace-dax-019-denied", ) assert binding.status == BINDING_STATUS_ACTIVE assert await _effect_count(db_session) == 0 resource_owner_result = await DataAccessService().revoke_data_access_binding( db_session, actor=resource_owner, binding_id=binding.id, reason="resource_owner_revoke", trace_id="trace-dax-019-resource-owner", ) assert resource_owner_result.authorized_as == "resource_owner" assert binding.status == BINDING_STATUS_REVOKED_PENDING second_resource = await _add_resource( db_session, owner_person_id=resource_owner.id, dataset_id="people", ) admin_binding = await _add_binding( db_session, resource=second_resource, group=group, entitlement=entitlement, ) admin_result = await DataAccessService().revoke_data_access_binding( db_session, actor=admin, binding_id=admin_binding.id, reason="admin_revoke", trace_id="trace-dax-019-admin", ) assert admin_result.authorized_as == "global_admin" assert admin_binding.status == BINDING_STATUS_REVOKED_PENDING legacy_admin = await _add_person(db_session, email="legacy-admin@example.com") legacy_admin.__dict__["_role_names_override"] = ["it_admin"] third_resource = await _add_resource( db_session, owner_person_id=resource_owner.id, dataset_id="ops", ) legacy_admin_binding = await _add_binding( db_session, resource=third_resource, group=group, entitlement=entitlement, ) legacy_admin_result = await DataAccessService().revoke_data_access_binding( db_session, actor=legacy_admin, binding_id=legacy_admin_binding.id, reason="legacy_admin_revoke", trace_id="trace-dax-019-it-admin", ) assert legacy_admin_result.authorized_as == "global_admin" assert legacy_admin_binding.status == BINDING_STATUS_REVOKED_PENDING async def test_resource_owner_revoke_fails_when_owner_authority_unresolved( db_session, ) -> None: resource_owner = await _add_person( db_session, email="fallback-owner@example.com", ) group_owner = await _add_person(db_session, email="group-owner@example.com") entitlement = await _add_entitlement(db_session) resource = await _add_resource( db_session, owner_person_id=resource_owner.id, owner_resolution_status=OWNER_RESOLUTION_FALLBACK, ) group = await _add_group( db_session, owner_person_id=group_owner.id, ) binding = await _add_binding( db_session, resource=resource, group=group, entitlement=entitlement, ) with pytest.raises(DataAccessRevokeAuthorizationError): await DataAccessService().revoke_data_access_binding( db_session, actor=resource_owner, binding_id=binding.id, reason="fallback_owner_must_not_authorize", trace_id="trace-dax-019-fallback-owner", ) assert binding.status == BINDING_STATUS_ACTIVE assert await _effect_count(db_session) == 0 async def test_revoke_absent_effect_uses_grant_time_target_locator( db_session, employee_user, ) -> None: group_owner, _ = employee_user entitlement = await _add_entitlement(db_session) resource = await _add_resource( db_session, owner_person_id=group_owner.id, ) group = await _add_group( db_session, owner_person_id=group_owner.id, ) binding = await _add_binding( db_session, resource=resource, group=group, entitlement=entitlement, ) await _add_effect( db_session, binding=binding, resource=resource, entitlement=entitlement, target_project_id="helix-data-original", target_dataset_id="finance_original", ) resource.project_id = "helix-data-renamed" resource.dataset_id = "finance_renamed" resource.source_version = "bootstrap-2026-06-11" resource.source_etag = "etag-renamed" await db_session.flush() await DataAccessService().revoke_data_access_binding( db_session, actor=group_owner, binding_id=binding.id, reason="group_owner_revoke", trace_id="trace-dax-019-grant-target", ) absent = await db_session.scalar( select(DataAccessCompiledEffect).where( DataAccessCompiledEffect.desired_state == EFFECT_DESIRED_ABSENT, ), ) assert absent is not None assert absent.target_project_id == "helix-data-original" assert absent.target_dataset_id == "finance_original" assert absent.resource_source_version_snapshot == "bootstrap-2026-06-10" assert absent.resource_source_etag_snapshot == "etag-finance" async def test_admin_confirmed_resource_removal_revokes_bindings_without_provider_io( db_session, admin_user, ) -> None: admin, _ = admin_user entitlement = await _add_entitlement(db_session) resource = await _add_resource( db_session, owner_person_id=admin.id, dataset_id="removed_finance", ) group = await _add_group( db_session, owner_person_id=admin.id, email="removed-finance-readers@example.com", ) binding = await _add_binding( db_session, resource=resource, group=group, entitlement=entitlement, ) present = await _add_effect( db_session, binding=binding, resource=resource, entitlement=entitlement, target_project_id="helix-data-original", target_dataset_id="removed_finance_original", ) result = await DataAccessService().confirm_data_access_resource_removed( db_session, resource_id=resource.id, reason=DATA_ACCESS_RESOURCE_REMOVAL_REASON_ADMIN_REMOVED, actor_type="user", actor_user_id=admin.id, trace_id="trace-dax-048-admin-removal", ) assert result.status == RESOURCE_STATUS_DELETED assert result.reason == DATA_ACCESS_RESOURCE_REMOVAL_REASON_ADMIN_REMOVED assert result.authorized_as == "user" assert result.revoked_pending_count == 1 assert result.effect_count == 1 await db_session.refresh(resource) await db_session.refresh(binding) assert resource.status == RESOURCE_STATUS_DELETED assert resource.requestable is False assert binding.status == BINDING_STATUS_REVOKED_PENDING assert binding.revoke_reason == DATA_ACCESS_RESOURCE_REMOVAL_REASON_ADMIN_REMOVED assert binding.revoked_at is not None states = set( await db_session.scalars( select(DataAccessCompiledEffect.desired_state).where( DataAccessCompiledEffect.binding_id == binding.id, ), ), ) assert states == {EFFECT_DESIRED_PRESENT, EFFECT_DESIRED_ABSENT} absent = await db_session.scalar( select(DataAccessCompiledEffect).where( DataAccessCompiledEffect.binding_id == binding.id, DataAccessCompiledEffect.desired_state == EFFECT_DESIRED_ABSENT, ), ) assert absent is not None assert absent.target_project_id == present.target_project_id assert absent.target_dataset_id == present.target_dataset_id assert absent.resource_source_version_snapshot == present.resource_source_version_snapshot assert absent.resource_source_etag_snapshot == present.resource_source_etag_snapshot assert absent.managed_reason == EFFECT_MANAGED_REASON_REVOKE_REQUEST assert absent.status == EFFECT_STATUS_PENDING_RECONCILE assert await _approval_request_count(db_session) == 0 audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.resource.removal_confirmed", ), ) assert audit is not None assert audit.actor_user_id == admin.id assert audit.detail["reason"] == DATA_ACCESS_RESOURCE_REMOVAL_REASON_ADMIN_REMOVED assert audit.detail["revoked_pending_count"] == 1 assert audit.detail["effect_count"] == 1 assert audit.detail["provider_mutation_attempted"] is False replay = await DataAccessService().confirm_data_access_resource_removed( db_session, resource_id=resource.id, reason=DATA_ACCESS_RESOURCE_REMOVAL_REASON_ADMIN_REMOVED, actor_type="user", actor_user_id=admin.id, trace_id="trace-dax-048-admin-removal-replay", ) assert replay.revoked_pending_count == 0 assert replay.effect_count == 0 assert replay.already_pending_count == 1 assert await _effect_count(db_session) == 2 async def test_system_deleted_group_revokes_bindings_idempotently_without_feishu( db_session, employee_user, ) -> None: owner, _ = employee_user entitlement = await _add_entitlement(db_session) resources = [ await _add_resource( db_session, owner_person_id=owner.id, dataset_id="finance", ), await _add_resource( db_session, owner_person_id=owner.id, dataset_id="people", ), ] group = await _add_group( db_session, owner_person_id=owner.id, ) bindings = [ await _add_binding( db_session, resource=resource, group=group, entitlement=entitlement, ) for resource in resources ] group.status = "deleted" await db_session.flush() first = await DataAccessService().revoke_invalid_google_group_bindings( db_session, email_group_id=group.id, trace_id="trace-dax-019-system-delete", ) second = await DataAccessService().revoke_invalid_google_group_bindings( db_session, email_group_id=group.id, trace_id="trace-dax-019-system-delete", ) assert first.revoked_pending_count == 2 assert first.effect_count == 2 assert first.authorized_as == "system" assert second.revoked_pending_count == 0 assert second.effect_count == 0 assert {binding.status for binding in bindings} == { BINDING_STATUS_REVOKED_PENDING, } assert {binding.revoke_reason for binding in bindings} == { "group_deleted_without_bridge", } effects = ( await db_session.scalars( select(DataAccessCompiledEffect).order_by( DataAccessCompiledEffect.target_dataset_id, ), ) ).all() assert [effect.desired_state for effect in effects] == [ EFFECT_DESIRED_ABSENT, EFFECT_DESIRED_ABSENT, ] assert {effect.provider_member for effect in effects} == { "group:analytics-readers@example.com", } assert await _approval_request_count(db_session) == 0 audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.binding.system_revoked_group", ), ) assert audit is not None assert audit.trace_id == AuditService._coerce_trace_id( "trace-dax-019-system-delete", ) assert audit.actor_type == "system" assert audit.detail is not None assert audit.detail["revoked_pending_count"] == 2 assert audit.detail["reason"] == "group_deleted_without_bridge" compile_audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.compile.output", AuditLog.resource_id == str(group.id), ), ) assert compile_audit is not None assert compile_audit.trace_id == audit.trace_id assert compile_audit.detail is not None assert compile_audit.detail["phase"] == "system_group_revoke" assert compile_audit.detail["effect_count"] == 2 assert compile_audit.detail["revoked_pending_count"] == 2 assert compile_audit.detail["reason"] == "group_deleted_without_bridge" assert { effect["effect_id"] for effect in compile_audit.detail["effects"] } == {str(effect.id) for effect in effects} assert { effect["desired_state"] for effect in compile_audit.detail["effects"] } == {EFFECT_DESIRED_ABSENT} async def test_system_deleted_group_with_ambiguous_bridge_does_not_revoke( db_session, employee_user, ) -> None: owner, _ = employee_user entitlement = await _add_entitlement(db_session) resource = await _add_resource(db_session, owner_person_id=owner.id) deleted_group = await _add_group( db_session, owner_person_id=owner.id, email="analytics-old@example.com", status="deleted", google_group_id="google-group-analytics", ) await _add_group( db_session, owner_person_id=owner.id, email="analytics-new-a@example.com", status="active", google_group_id="google-group-analytics", ) await _add_group( db_session, owner_person_id=owner.id, email="analytics-new-b@example.com", status="active", google_group_id="google-group-analytics", ) binding = await _add_binding( db_session, resource=resource, group=deleted_group, entitlement=entitlement, email_snapshot="analytics-old@example.com", ) result = await DataAccessService().revoke_invalid_google_group_bindings( db_session, email_group_id=deleted_group.id, trace_id="trace-dax-019-ambiguous-bridge", ) assert result.status == "blocked" assert result.reason == "group_rename_bridge_ambiguous" assert result.revoked_pending_count == 0 assert result.effect_count == 0 assert binding.status == BINDING_STATUS_ACTIVE assert await _effect_count(db_session) == 0 audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.binding.system_revoke_blocked", ), ) assert audit is not None assert audit.trace_id == AuditService._coerce_trace_id( "trace-dax-019-ambiguous-bridge", ) assert audit.status == "failure" assert audit.detail is not None assert audit.detail["reason"] == "group_rename_bridge_ambiguous" async def test_system_rename_bridge_compiles_absent_and_present_effects( db_session, employee_user, ) -> None: owner, _ = employee_user entitlement = await _add_entitlement(db_session) resource = await _add_resource(db_session, owner_person_id=owner.id) old_group = await _add_group( db_session, owner_person_id=owner.id, email="analytics-old@example.com", status="deleted", google_group_id="google-group-analytics", ) new_group = await _add_group( db_session, owner_person_id=owner.id, email="analytics-new@example.com", status="active", google_group_id="google-group-analytics", ) binding = await _add_binding( db_session, resource=resource, group=old_group, entitlement=entitlement, email_snapshot="analytics-old@example.com", ) result = await DataAccessService().revoke_invalid_google_group_bindings( db_session, email_group_id=old_group.id, trace_id="trace-dax-019-rename", ) assert result.renamed_binding_count == 1 assert result.revoked_pending_count == 0 assert result.effect_count == 2 assert binding.status == BINDING_STATUS_ACTIVE assert binding.principal_email_group_id == new_group.id assert binding.principal_email_snapshot == "analytics-new@example.com" assert binding.revoke_reason is None assert binding.revoked_at is None effects = ( await db_session.scalars( select(DataAccessCompiledEffect).order_by( DataAccessCompiledEffect.desired_state, ), ) ).all() assert {effect.desired_state for effect in effects} == { EFFECT_DESIRED_ABSENT, EFFECT_DESIRED_PRESENT, } assert {effect.managed_reason for effect in effects} == { EFFECT_MANAGED_REASON_GROUP_RENAME_BRIDGE, } assert { (effect.desired_state, effect.provider_member) for effect in effects } == { (EFFECT_DESIRED_ABSENT, "group:analytics-old@example.com"), (EFFECT_DESIRED_PRESENT, "group:analytics-new@example.com"), } assert {effect.principal_email_group_id for effect in effects} == { new_group.id, } assert await _approval_request_count(db_session) == 0 compile_audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.compile.output", AuditLog.resource_id == str(old_group.id), ), ) assert compile_audit is not None assert compile_audit.trace_id == AuditService._coerce_trace_id( "trace-dax-019-rename", ) assert compile_audit.detail is not None assert compile_audit.detail["phase"] == "group_rename_bridge" assert compile_audit.detail["effect_count"] == 2 assert compile_audit.detail["renamed_binding_count"] == 1 assert compile_audit.detail["superseded_binding_count"] == 0 assert { (effect["desired_state"], effect["provider_member"]) for effect in compile_audit.detail["effects"] } == { (EFFECT_DESIRED_ABSENT, "group:analytics-old@example.com"), (EFFECT_DESIRED_PRESENT, "group:analytics-new@example.com"), } async def test_system_rename_bridge_supersedes_old_binding_on_new_binding_conflict( db_session, employee_user, ) -> None: owner, _ = employee_user entitlement = await _add_entitlement(db_session) resource = await _add_resource(db_session, owner_person_id=owner.id) old_group = await _add_group( db_session, owner_person_id=owner.id, email="analytics-old@example.com", status="deleted", google_group_id="google-group-analytics", ) new_group = await _add_group( db_session, owner_person_id=owner.id, email="analytics-new@example.com", status="active", google_group_id="google-group-analytics", ) old_binding = await _add_binding( db_session, resource=resource, group=old_group, entitlement=entitlement, email_snapshot="analytics-old@example.com", ) await _add_effect( db_session, binding=old_binding, resource=resource, entitlement=entitlement, ) new_binding = await _add_binding( db_session, resource=resource, group=new_group, entitlement=entitlement, email_snapshot="analytics-new@example.com", ) result = await DataAccessService().revoke_invalid_google_group_bindings( db_session, email_group_id=old_group.id, trace_id="trace-dax-019-rename-conflict", ) assert result.renamed_binding_count == 0 assert result.revoked_pending_count == 1 assert result.effect_count == 2 assert old_binding.status == BINDING_STATUS_REVOKED_PENDING assert old_binding.revoke_reason == "group_rename_bridge_superseded" assert old_binding.principal_email_group_id == old_group.id assert new_binding.status == BINDING_STATUS_ACTIVE bridge_effects = ( await db_session.scalars( select(DataAccessCompiledEffect) .where( DataAccessCompiledEffect.managed_reason == EFFECT_MANAGED_REASON_GROUP_RENAME_BRIDGE, ) .order_by(DataAccessCompiledEffect.desired_state) ) ).all() assert { (effect.binding_id, effect.desired_state, effect.provider_member) for effect in bridge_effects } == { ( old_binding.id, EFFECT_DESIRED_ABSENT, "group:analytics-old@example.com", ), ( new_binding.id, EFFECT_DESIRED_PRESENT, "group:analytics-new@example.com", ), } async def test_system_rename_bridge_does_not_regrant_revoked_pending_new_binding( db_session, employee_user, ) -> None: owner, _ = employee_user entitlement = await _add_entitlement(db_session) resource = await _add_resource(db_session, owner_person_id=owner.id) old_group = await _add_group( db_session, owner_person_id=owner.id, email="analytics-old@example.com", status="deleted", google_group_id="google-group-analytics", ) new_group = await _add_group( db_session, owner_person_id=owner.id, email="analytics-new@example.com", status="active", google_group_id="google-group-analytics", ) old_binding = await _add_binding( db_session, resource=resource, group=old_group, entitlement=entitlement, email_snapshot="analytics-old@example.com", ) await _add_effect( db_session, binding=old_binding, resource=resource, entitlement=entitlement, ) new_binding = await _add_binding( db_session, resource=resource, group=new_group, entitlement=entitlement, email_snapshot="analytics-new@example.com", ) new_binding.status = BINDING_STATUS_REVOKED_PENDING new_binding.revoke_reason = "existing_new_group_revoke" await _add_effect( db_session, binding=new_binding, resource=resource, entitlement=entitlement, desired_state=EFFECT_DESIRED_ABSENT, managed_reason=EFFECT_MANAGED_REASON_REVOKE_REQUEST, provider_member="group:analytics-new@example.com", ) await db_session.flush() result = await DataAccessService().revoke_invalid_google_group_bindings( db_session, email_group_id=old_group.id, trace_id="trace-dax-019-rename-revoked-conflict", ) assert result.renamed_binding_count == 0 assert result.revoked_pending_count == 1 assert result.effect_count == 1 assert old_binding.status == BINDING_STATUS_REVOKED_PENDING assert old_binding.revoke_reason == "group_rename_bridge_superseded" assert new_binding.status == BINDING_STATUS_REVOKED_PENDING assert new_binding.revoke_reason == "existing_new_group_revoke" bridge_effects = ( await db_session.scalars( select(DataAccessCompiledEffect) .where( DataAccessCompiledEffect.managed_reason == EFFECT_MANAGED_REASON_GROUP_RENAME_BRIDGE, ) .order_by(DataAccessCompiledEffect.desired_state) ) ).all() assert { (effect.binding_id, effect.desired_state, effect.provider_member) for effect in bridge_effects } == { ( old_binding.id, EFFECT_DESIRED_ABSENT, "group:analytics-old@example.com", ), } async def test_system_invalid_group_without_google_id_revokes_binding( db_session, employee_user, ) -> None: owner, _ = employee_user entitlement = await _add_entitlement(db_session) resource = await _add_resource(db_session, owner_person_id=owner.id) group = await _add_group( db_session, owner_person_id=owner.id, google_group_id=None, ) binding = await _add_binding( db_session, resource=resource, group=group, entitlement=entitlement, ) result = await DataAccessService().revoke_invalid_google_group_bindings( db_session, email_group_id=group.id, trace_id="trace-dax-019-invalid", ) assert result.revoked_pending_count == 1 assert binding.status == BINDING_STATUS_REVOKED_PENDING assert binding.revoke_reason == "group_invalid" effect = await db_session.scalar(select(DataAccessCompiledEffect)) assert effect is not None assert effect.desired_state == EFFECT_DESIRED_ABSENT assert effect.provider_member == "group:analytics-readers@example.com" async def test_system_active_duplicate_google_group_id_revokes_as_invalid_signal( db_session, employee_user, ) -> None: owner, _ = employee_user entitlement = await _add_entitlement(db_session) resource = await _add_resource(db_session, owner_person_id=owner.id) group = await _add_group( db_session, owner_person_id=owner.id, email="analytics-a@example.com", google_group_id="google-group-analytics", ) await _add_group( db_session, owner_person_id=owner.id, email="analytics-b@example.com", google_group_id="google-group-analytics", ) binding = await _add_binding( db_session, resource=resource, group=group, entitlement=entitlement, email_snapshot="analytics-a@example.com", ) result = await DataAccessService().revoke_invalid_google_group_bindings( db_session, email_group_id=group.id, trace_id="trace-dax-019-duplicate-active", ) assert result.revoked_pending_count == 1 assert result.reason == "group_duplicate_active_google_group_id" assert binding.status == BINDING_STATUS_REVOKED_PENDING assert binding.revoke_reason == "group_duplicate_active_google_group_id" effect = await db_session.scalar(select(DataAccessCompiledEffect)) assert effect is not None assert effect.desired_state == EFFECT_DESIRED_ABSENT assert effect.provider_member == "group:analytics-a@example.com" audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.binding.system_revoked_group", ), ) assert audit is not None assert audit.trace_id == AuditService._coerce_trace_id( "trace-dax-019-duplicate-active", ) assert audit.detail is not None assert audit.detail["reason"] == "group_duplicate_active_google_group_id" def test_revoke_service_does_not_mutate_bigquery_or_feishu_directly() -> None: checked_sources = [ inspect.getsource(DataAccessService.revoke_data_access_binding), inspect.getsource(DataAccessService.revoke_invalid_google_group_bindings), inspect.getsource(DataAccessService.confirm_data_access_resource_removed), inspect.getsource(DataAccessService._bridge_renamed_group_bindings), ] forbidden_tokens = ( "google.cloud", "bigquery.Client", "set_iam_policy", "setIamPolicy", "providers.gcp", "lifecycle_service", "create_instance(", "cancel_instance(", ) for text in checked_sources: for token in forbidden_tokens: assert token not in text, f"revoke path must not use {token}" async def test_revoke_missing_binding_returns_404( db_session, employee_user, ) -> None: actor, _ = employee_user with pytest.raises(AppException) as exc: await DataAccessService().revoke_data_access_binding( db_session, actor=actor, binding_id=uuid.uuid4(), reason="missing", trace_id="trace-dax-019-missing", ) assert exc.value.code == "DATA_ACCESS_BINDING_NOT_FOUND"