from __future__ import annotations import asyncio import uuid from pathlib import Path from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.audit.models import AuditLog from app.domains.data_access import constants as c 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_OBSERVED, EFFECT_STATUS_PENDING_RECONCILE, EFFECT_STATUS_RETRYABLE, ENTITLEMENT_STATUS_ACTIVE, ENTITLEMENT_VERSION_V1, OWNER_RESOLUTION_RESOLVED, OWNER_SOURCE_BOOTSTRAP, RECONCILE_STATUS_APPLIED, RECONCILE_STATUS_OBSERVED, RECONCILE_STATUS_RETRYABLE, REQUEST_ITEM_STATUS_APPROVED, REQUEST_STATUS_APPROVED, RESOURCE_STATUS_ACTIVE, DataAccessBinding, DataAccessCompiledEffect, DataAccessEffectRun, DataAccessEntitlement, DataAccessPrincipal, DataAccessRequest, DataAccessRequestItem, DataAccessResource, ) from app.domains.data_access.reconciler import ( APPLY_ABSENT_OUTCOME_ALREADY_ABSENT, APPLY_ABSENT_OUTCOME_APPLIED, APPLY_ABSENT_OUTCOME_CONFLICT, APPLY_ABSENT_OUTCOME_ERROR, APPLY_GATE_REASON_ALLOWED, APPLY_GATE_REASON_DISABLED, APPLY_GATE_REASON_INVALID_CONFIG, APPLY_PRESENT_OUTCOME_ALREADY_APPLIED, APPLY_PRESENT_OUTCOME_APPLIED, APPLY_PRESENT_OUTCOME_CONFLICT, APPLY_PRESENT_OUTCOME_ERROR, OBSERVE_OUTCOME_ERROR, OBSERVE_OUTCOME_UNCHANGED, OBSERVE_OUTCOME_WOULD_ADD, OBSERVE_OUTCOME_WOULD_REMOVE, apply_absent_reconcile_compiled_effects, apply_present_reconcile_compiled_effects, observe_reconcile_compiled_effects, resolve_apply_reconcile_gate, ) from app.domains.data_access.service import DATA_ACCESS_COMPILER_VERSION, DataAccessService from app.domains.email_group.models import EmailGroup from app.integrations.driver_base import ( DataAccessDatasetIamBinding, DataAccessDatasetIamPolicy, DataAccessDatasetIamPolicyPatch, DataAccessDatasetIamUpdateResult, DataAccessDriver, ) from app.integrations.models import Integration class FakeObserveDriver(DataAccessDriver): provider = c.PROVIDER_BIGQUERY def __init__( self, *, policies: dict[tuple[str, str], DataAccessDatasetIamPolicy] | None = None, table_policies: dict[ tuple[str, str, str], DataAccessDatasetIamPolicy ] | None = None, error: Exception | None = None, get_started: asyncio.Event | None = None, wait_before_return: asyncio.Event | None = None, ) -> None: super().__init__( integration_id="fake-data-access-integration", config={}, ) self.policies = policies or {} self.table_policies = table_policies or {} self.error = error self.get_started = get_started self.wait_before_return = wait_before_return self.get_calls: list[tuple[str, str]] = [] self.table_get_calls: list[tuple[str, str, str]] = [] self.set_calls: list[tuple[Any, ...]] = [] self.table_set_calls: list[tuple[Any, ...]] = [] self.forbidden_calls: list[str] = [] async def get_dataset_iam( self, project_id: str, dataset_id: str, ) -> DataAccessDatasetIamPolicy: self.get_calls.append((project_id, dataset_id)) if self.get_started is not None: self.get_started.set() if self.wait_before_return is not None: await self.wait_before_return.wait() if self.error is not None: raise self.error return self.policies[(project_id, dataset_id)] async def get_table_iam( self, project_id: str, dataset_id: str, table_id: str, ) -> DataAccessDatasetIamPolicy: self.table_get_calls.append((project_id, dataset_id, table_id)) return self.table_policies[(project_id, dataset_id, table_id)] async def set_dataset_iam( self, project_id: str, dataset_id: str, expected_etag: str, policy_patch: DataAccessDatasetIamPolicyPatch, ): self.set_calls.append((project_id, dataset_id, expected_etag, policy_patch)) raise AssertionError("observe-mode reconciler must not call set_dataset_iam") async def list_datasets(self, *args: Any, **kwargs: Any): self.forbidden_calls.append("list_datasets") raise AssertionError("observe-mode reconciler must not list datasets") async def list_all_datasets(self, *args: Any, **kwargs: Any): self.forbidden_calls.append("list_all_datasets") raise AssertionError("observe-mode reconciler must not list all datasets") async def list_resources(self, *args: Any, **kwargs: Any): self.forbidden_calls.append("list_resources") raise AssertionError("observe-mode reconciler must not list resources") async def list_all_resources(self, *args: Any, **kwargs: Any): self.forbidden_calls.append("list_all_resources") raise AssertionError("observe-mode reconciler must not list all resources") async def scan_resources(self, *args: Any, **kwargs: Any): self.forbidden_calls.append("scan_resources") raise AssertionError("observe-mode reconciler must not scan resources") async def get_project_iam(self, *args: Any, **kwargs: Any): self.forbidden_calls.append("get_project_iam") raise AssertionError("observe-mode reconciler must not read project IAM") async def set_project_iam(self, *args: Any, **kwargs: Any): self.forbidden_calls.append("set_project_iam") raise AssertionError("observe-mode reconciler must not write project IAM") async def get_project_policy(self, *args: Any, **kwargs: Any): self.forbidden_calls.append("get_project_policy") raise AssertionError("observe-mode reconciler must not read project policy") async def set_project_policy(self, *args: Any, **kwargs: Any): self.forbidden_calls.append("set_project_policy") raise AssertionError("observe-mode reconciler must not write project policy") async def grant_access(self, *args: Any, **kwargs: Any): self.forbidden_calls.append("grant_access") raise AssertionError("observe-mode reconciler must not use cloud grants") async def revoke_access(self, *args: Any, **kwargs: Any): self.forbidden_calls.append("revoke_access") raise AssertionError("observe-mode reconciler must not use cloud grants") class FakeApplyDriver(FakeObserveDriver): def __init__( self, *, policies: dict[tuple[str, str], DataAccessDatasetIamPolicy] | None = None, table_policies: dict[ tuple[str, str, str], DataAccessDatasetIamPolicy ] | None = None, apply_results: list[DataAccessDatasetIamUpdateResult] | None = None, error: Exception | None = None, ) -> None: super().__init__( policies=policies, table_policies=table_policies, error=error, ) self.apply_results = list(apply_results or []) 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", ) async def set_table_iam( self, project_id: str, dataset_id: str, table_id: str, expected_etag: str | None, policy_patch: DataAccessDatasetIamPolicyPatch, ) -> DataAccessDatasetIamUpdateResult: self.table_set_calls.append(( project_id, dataset_id, table_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", ) async def _integration_id( db_session, *, slug_prefix: str = "dax-observe", config_public: dict[str, Any] | None = None, kind: str = c.DOMAIN_DATA_ACCESS, provider: str = c.PROVIDER_BIGQUERY, enabled: bool = True, ) -> str: row = Integration( kind=kind, provider=provider, name="Data Access BigQuery", slug=f"{slug_prefix}-{uuid.uuid4().hex[:8]}", enabled=enabled, is_primary=False, config_public=config_public, ) 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, project_id: str = "helix-data-prod", dataset_id: str = "finance", provider: str = c.PROVIDER_BIGQUERY, resource_type: str = c.RESOURCE_TYPE_DATASET, table_id: str | None = None, owner_person_id: uuid.UUID, ) -> DataAccessResource: return DataAccessResource( integration_id=integration_id, provider=provider, resource_type=resource_type, project_id=project_id, dataset_id=dataset_id, table_id=table_id, table_type="TABLE" if table_id else None, display_name=f"{project_id}.{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}", ) def _group(email: str) -> EmailGroup: return EmailGroup( email=email, name=email.split("@", 1)[0], status="active", google_group_id=f"google-group-{uuid.uuid4()}", ) def _binding( *, resource: DataAccessResource, group: EmailGroup, entitlement: DataAccessEntitlement, request: DataAccessRequest | None = None, item: DataAccessRequestItem | 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=group.email, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, source_request_id=request.id if request else None, source_request_item_id=item.id if item else None, status=status, ) def _effect( *, binding: DataAccessBinding, resource: DataAccessResource, group: EmailGroup, entitlement: DataAccessEntitlement, desired_state: str = EFFECT_DESIRED_PRESENT, status: str = EFFECT_STATUS_PENDING_RECONCILE, ) -> 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=resource.resource_type, target_project_id=resource.project_id, target_dataset_id=resource.dataset_id, target_table_id=resource.table_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=( EFFECT_MANAGED_REASON_GRANT_REQUEST if desired_state == EFFECT_DESIRED_PRESENT else EFFECT_MANAGED_REASON_REVOKE_REQUEST ), already_present_at_first_apply=False, status=status, ) async def _seed_effect( db_session, *, resource: DataAccessResource, group: EmailGroup, entitlement: DataAccessEntitlement, request: DataAccessRequest | None = None, item: DataAccessRequestItem | None = None, desired_state: str = EFFECT_DESIRED_PRESENT, binding_status: str = BINDING_STATUS_ACTIVE, effect_status: str = EFFECT_STATUS_PENDING_RECONCILE, ) -> DataAccessCompiledEffect: binding = _binding( resource=resource, group=group, entitlement=entitlement, request=request, item=item, status=binding_status, ) db_session.add(binding) await db_session.flush() effect = _effect( binding=binding, resource=resource, group=group, entitlement=entitlement, desired_state=desired_state, status=effect_status, ) db_session.add(effect) await db_session.flush() return effect def _request( *, requester_id: uuid.UUID, resource: DataAccessResource, group: EmailGroup, entitlement: DataAccessEntitlement, ) -> DataAccessRequest: return 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=f"dax-feishu-{uuid.uuid4().hex[:8]}", status=REQUEST_STATUS_APPROVED, 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=uuid.UUID( str(resource.owner_person_id), ), status=REQUEST_ITEM_STATUS_APPROVED, ), ], ) async def test_apply_gate_disabled_preserves_pending_effects( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-disabled", config_public={ "apply_mode_enabled": False, }, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("apply-disabled@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, ) result = await resolve_apply_reconcile_gate( db_session, integration_id=integration_id, trace_id="dax-032-disabled-gate-test", ) assert result.apply_mode_enabled is False assert result.selected_effect_count == 1 assert result.allowed_effect_count == 0 assert result.blocked_effect_count == 1 assert result.decisions[0].allowed is False assert result.decisions[0].reason == APPLY_GATE_REASON_DISABLED assert effect.status == EFFECT_STATUS_PENDING_RECONCILE assert effect.last_provider_outcome is None assert effect.last_provider_etag is None assert effect.last_error is None audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.apply_gate") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.status == "success" assert audit.detail["mode"] == "apply_gate" assert audit.detail["provider_mutation_attempted"] is False assert audit.detail["apply_mode_enabled"] is False assert audit.detail["allowed_effect_count"] == 0 assert audit.detail["blocked_effect_count"] == 1 assert audit.detail["decisions"] == [{ "effect_id": str(effect.id), "resource_id": str(resource.id), "target_project_id": "helix-data-prod", "target_dataset_id": "finance", "provider_member": "group:apply-disabled@example.com", "provider_role": c.BIGQUERY_DATA_VIEWER_ROLE, "desired_state": EFFECT_DESIRED_PRESENT, "locator": "helix-data-prod.finance", "allowed": False, "reason": APPLY_GATE_REASON_DISABLED, }] async def test_apply_gate_allows_selected_effects_when_apply_mode_enabled( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-canary", config_public={ "apply_mode_enabled": "yes", }, ) entitlement = _entitlement() finance_resource = _resource( integration_id=integration_id, dataset_id="finance", owner_person_id=owner.id, ) hr_resource = _resource( integration_id=integration_id, dataset_id="hr", owner_person_id=owner.id, ) finance_group = _group("finance-apply@example.com") hr_group = _group("hr-apply@example.com") db_session.add_all([ entitlement, finance_resource, hr_resource, finance_group, hr_group, ]) await db_session.flush() finance_effect = await _seed_effect( db_session, resource=finance_resource, group=finance_group, entitlement=entitlement, ) hr_effect = await _seed_effect( db_session, resource=hr_resource, group=hr_group, entitlement=entitlement, ) result = await resolve_apply_reconcile_gate( db_session, integration_id=integration_id, resource_ids=[finance_resource.id, hr_resource.id], trace_id="dax-032-canary-gate-test", ) assert result.apply_mode_enabled is True assert result.selected_effect_count == 2 assert result.allowed_effect_count == 2 assert result.blocked_effect_count == 0 decisions = {decision.locator: decision for decision in result.decisions} assert decisions["helix-data-prod.finance"].effect_id == finance_effect.id assert decisions["helix-data-prod.finance"].allowed is True assert decisions["helix-data-prod.finance"].reason == APPLY_GATE_REASON_ALLOWED assert decisions["helix-data-prod.hr"].effect_id == hr_effect.id assert decisions["helix-data-prod.hr"].allowed is True assert decisions["helix-data-prod.hr"].reason == APPLY_GATE_REASON_ALLOWED assert finance_effect.status == EFFECT_STATUS_PENDING_RECONCILE assert hr_effect.status == EFFECT_STATUS_PENDING_RECONCILE audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.apply_gate") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.status == "success" assert audit.detail["apply_mode_enabled"] is True assert audit.detail["allowed_effect_count"] == 2 assert audit.detail["blocked_effect_count"] == 0 assert {row["locator"]: row["reason"] for row in audit.detail["decisions"]} == { "helix-data-prod.finance": APPLY_GATE_REASON_ALLOWED, "helix-data-prod.hr": APPLY_GATE_REASON_ALLOWED, } async def test_apply_gate_invalid_config_fails_closed( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-invalid", config_public=["not-an-object"], ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("apply-invalid@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, ) result = await resolve_apply_reconcile_gate( db_session, integration_id=integration_id, trace_id="dax-032-invalid-gate-test", ) assert result.config_error == ( "BigQuery DAX apply gate config must be an object" ) assert result.apply_mode_enabled is False assert result.allowed_effect_count == 0 assert result.blocked_effect_count == 1 assert result.decisions[0].reason == APPLY_GATE_REASON_INVALID_CONFIG assert effect.status == EFFECT_STATUS_PENDING_RECONCILE assert effect.last_provider_outcome is None audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.apply_gate") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.status == "failure" assert audit.detail["config_error"] == result.config_error assert audit.detail["allowed_effect_count"] == 0 assert audit.detail["blocked_effect_count"] == 1 assert audit.detail["decisions"][0]["reason"] == APPLY_GATE_REASON_INVALID_CONFIG async def test_apply_gate_requires_enabled_data_access_bigquery_integration( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-disabled-integration", config_public={ "apply_mode_enabled": True, }, enabled=False, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("apply-disabled-integration@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, ) result = await resolve_apply_reconcile_gate( db_session, integration_id=integration_id, trace_id="dax-032-disabled-integration-test", ) assert result.config_error == ( "Data-access apply gate requires an enabled data_access/bigquery integration" ) assert result.apply_mode_enabled is False assert result.allowed_effect_count == 0 assert result.blocked_effect_count == 1 assert result.decisions[0].effect_id == effect.id assert result.decisions[0].reason == APPLY_GATE_REASON_INVALID_CONFIG assert effect.status == EFFECT_STATUS_PENDING_RECONCILE assert effect.last_provider_outcome is None audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.apply_gate") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.status == "failure" assert audit.detail["config_error"] == result.config_error assert audit.detail["provider_mutation_attempted"] is False assert audit.detail["decisions"][0]["reason"] == APPLY_GATE_REASON_INVALID_CONFIG async def test_apply_present_canary_effect_writes_dataset_iam_and_marks_applied( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-present", config_public={ "apply_mode_enabled": True, }, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("canary-analytics@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() request = _request( requester_id=owner.id, resource=resource, group=group, entitlement=entitlement, ) db_session.add(request) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, request=request, item=request.items[0], ) driver = FakeApplyDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-apply-1", bindings=(), raw_policy={"etag": "provider-etag-apply-1"}, ), }, apply_results=[ DataAccessDatasetIamUpdateResult( ok=True, changed=True, etag="provider-etag-apply-2", ), ], ) result = await apply_present_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-033-apply-present-test", ) assert driver.get_calls == [("helix-data-prod", "finance")] assert len(driver.set_calls) == 1 project_id, dataset_id, expected_etag, patch = driver.set_calls[0] assert (project_id, dataset_id, expected_etag) == ( "helix-data-prod", "finance", "provider-etag-apply-1", ) assert patch == DataAccessDatasetIamPolicyPatch( add_bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=("group:canary-analytics@example.com",), ), ), remove_bindings=(), ) assert result.selected_effect_count == 1 assert result.selected_present_effect_count == 1 assert result.ignored_non_present_effect_count == 0 assert result.allowed_effect_count == 1 assert result.blocked_effect_count == 0 assert result.provider_write_count == 1 assert result.applied_count == 1 assert result.already_applied_count == 0 assert result.conflict_count == 0 assert result.error_count == 0 assert result.results[0].effect_id == effect.id assert result.results[0].outcome == APPLY_PRESENT_OUTCOME_APPLIED assert effect.status == EFFECT_STATUS_APPLIED assert effect.last_provider_outcome == APPLY_PRESENT_OUTCOME_APPLIED assert effect.last_provider_etag == "provider-etag-apply-2" assert effect.last_error is None binding = await db_session.get(DataAccessBinding, effect.binding_id) assert binding is not None assert binding.reconcile_status == RECONCILE_STATUS_APPLIED assert binding.last_reconcile_run_id == result.run_id assert request.reconcile_status == RECONCILE_STATUS_APPLIED assert request.last_reconcile_run_id == result.run_id run = await db_session.scalar( select(DataAccessEffectRun).where(DataAccessEffectRun.effect_id == effect.id), ) assert run is not None assert run.run_id == result.run_id assert run.mode == "apply_present" assert run.request_id == request.id assert run.effect_status == EFFECT_STATUS_APPLIED assert run.provider_outcome == APPLY_PRESENT_OUTCOME_APPLIED assert run.provider_etag == "provider-etag-apply-2" audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.apply_present") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.status == "success" assert audit.detail["provider_mutation_attempted"] is True assert audit.detail["provider_write_count"] == 1 assert audit.detail["applied_count"] == 1 assert audit.detail["gate_decisions"] == [{ "effect_id": str(effect.id), "resource_id": str(resource.id), "target_project_id": "helix-data-prod", "target_dataset_id": "finance", "provider_member": "group:canary-analytics@example.com", "provider_role": c.BIGQUERY_DATA_VIEWER_ROLE, "desired_state": EFFECT_DESIRED_PRESENT, "locator": "helix-data-prod.finance", "allowed": True, "reason": APPLY_GATE_REASON_ALLOWED, }] assert audit.detail["results"][0]["outcome"] == APPLY_PRESENT_OUTCOME_APPLIED async def test_apply_present_table_effect_writes_only_exact_table_iam( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-table", config_public={"apply_mode_enabled": True}, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, resource_type=c.RESOURCE_TYPE_TABLE, table_id="account_daily", ) group = _group("table-canary@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, ) driver = FakeApplyDriver( table_policies={ ("helix-data-prod", "finance", "account_daily"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", table_id="account_daily", etag="table-etag-1", bindings=(), raw_policy={"etag": "table-etag-1"}, ), }, ) result = await apply_present_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-apply-table-test", ) assert driver.get_calls == [] assert driver.set_calls == [] assert driver.table_get_calls == [ ("helix-data-prod", "finance", "account_daily"), ] assert len(driver.table_set_calls) == 1 project_id, dataset_id, table_id, expected_etag, patch = ( driver.table_set_calls[0] ) assert (project_id, dataset_id, table_id, expected_etag) == ( "helix-data-prod", "finance", "account_daily", "table-etag-1", ) assert patch.add_bindings[0].members == ("group:table-canary@example.com",) assert result.results[0].target_table_id == "account_daily" run = await db_session.scalar( select(DataAccessEffectRun).where(DataAccessEffectRun.effect_id == effect.id), ) assert run is not None assert run.target_table_id == "account_daily" async def test_apply_present_already_applied_is_idempotent( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-already-present", config_public={ "apply_mode_enabled": True, }, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("already-present@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, ) driver = FakeApplyDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-already", bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=("group:already-present@example.com",), ), ), raw_policy={"etag": "provider-etag-already"}, ), }, apply_results=[ DataAccessDatasetIamUpdateResult( ok=True, changed=False, etag="provider-etag-already", already_applied=True, ), ], ) result = await apply_present_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-033-already-present-test", ) assert result.applied_count == 0 assert result.already_applied_count == 1 assert result.error_count == 0 assert result.results[0].outcome == APPLY_PRESENT_OUTCOME_ALREADY_APPLIED assert effect.status == EFFECT_STATUS_APPLIED assert effect.already_present_at_first_apply is True assert effect.last_provider_outcome == APPLY_PRESENT_OUTCOME_ALREADY_APPLIED assert effect.last_provider_etag == "provider-etag-already" audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.apply_present") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.detail["results"][0]["already_present_at_first_apply"] is True async def test_apply_present_mixed_batch_marks_only_preexisting_effect_as_adopted( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-mixed-present", config_public={ "apply_mode_enabled": True, }, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) existing_group = _group("adopted-mixed@example.com") new_group = _group("new-mixed@example.com") db_session.add_all([entitlement, resource, existing_group, new_group]) await db_session.flush() existing_request = _request( requester_id=owner.id, resource=resource, group=existing_group, entitlement=entitlement, ) new_request = _request( requester_id=owner.id, resource=resource, group=new_group, entitlement=entitlement, ) db_session.add_all([existing_request, new_request]) await db_session.flush() existing_effect = await _seed_effect( db_session, resource=resource, group=existing_group, entitlement=entitlement, request=existing_request, item=existing_request.items[0], ) new_effect = await _seed_effect( db_session, resource=resource, group=new_group, entitlement=entitlement, request=new_request, item=new_request.items[0], ) driver = FakeApplyDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-mixed-1", bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=("group:adopted-mixed@example.com",), ), ), raw_policy={"etag": "provider-etag-mixed-1"}, ), }, apply_results=[ DataAccessDatasetIamUpdateResult( ok=True, changed=True, etag="provider-etag-mixed-2", ), ], ) result = await apply_present_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-035-mixed-adopt-present-test", ) assert len(driver.set_calls) == 1 project_id, dataset_id, expected_etag, patch = driver.set_calls[0] assert (project_id, dataset_id, expected_etag) == ( "helix-data-prod", "finance", "provider-etag-mixed-1", ) assert patch == DataAccessDatasetIamPolicyPatch( add_bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=( "group:adopted-mixed@example.com", "group:new-mixed@example.com", ), ), ), remove_bindings=(), ) assert result.provider_write_count == 1 assert result.applied_count == 1 assert result.already_applied_count == 1 results_by_effect_id = {row.effect_id: row for row in result.results} existing_result = results_by_effect_id[existing_effect.id] new_result = results_by_effect_id[new_effect.id] assert existing_result.outcome == APPLY_PRESENT_OUTCOME_ALREADY_APPLIED assert existing_result.already_present_at_first_apply is True assert new_result.outcome == APPLY_PRESENT_OUTCOME_APPLIED assert new_result.already_present_at_first_apply is False assert existing_effect.status == EFFECT_STATUS_APPLIED assert existing_effect.already_present_at_first_apply is True assert existing_effect.last_provider_outcome == ( APPLY_PRESENT_OUTCOME_ALREADY_APPLIED ) assert new_effect.status == EFFECT_STATUS_APPLIED assert new_effect.already_present_at_first_apply is False assert new_effect.last_provider_outcome == APPLY_PRESENT_OUTCOME_APPLIED audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.apply_present") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.detail["applied_count"] == 1 assert audit.detail["already_applied_count"] == 1 audit_results_by_effect_id = { uuid.UUID(row["effect_id"]): row for row in audit.detail["results"] } assert ( audit_results_by_effect_id[existing_effect.id][ "already_present_at_first_apply" ] is True ) assert ( audit_results_by_effect_id[new_effect.id][ "already_present_at_first_apply" ] is False ) async def test_apply_present_adopts_already_present_then_revoke_removes_exact_effect( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-adopt-present", config_public={ "apply_mode_enabled": True, }, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("adopted-analytics@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() request = _request( requester_id=owner.id, resource=resource, group=group, entitlement=entitlement, ) db_session.add(request) await db_session.flush() present_effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, request=request, item=request.items[0], ) present_driver = FakeApplyDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-adopt-1", bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=("group:adopted-analytics@example.com",), ), ), raw_policy={"etag": "provider-etag-adopt-1"}, ), }, apply_results=[ DataAccessDatasetIamUpdateResult( ok=True, changed=False, etag="provider-etag-adopt-1", already_applied=True, ), ], ) present_result = await apply_present_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=present_driver, trace_id="dax-035-adopt-present-test", ) assert len(present_driver.set_calls) == 1 _project_id, _dataset_id, _expected_etag, present_patch = ( present_driver.set_calls[0] ) assert present_patch == DataAccessDatasetIamPolicyPatch( add_bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=("group:adopted-analytics@example.com",), ), ), remove_bindings=(), ) assert present_result.applied_count == 0 assert present_result.already_applied_count == 1 assert present_result.results[0].outcome == ( APPLY_PRESENT_OUTCOME_ALREADY_APPLIED ) assert present_result.results[0].already_present_at_first_apply is True assert present_effect.status == EFFECT_STATUS_APPLIED assert present_effect.last_provider_outcome == ( APPLY_PRESENT_OUTCOME_ALREADY_APPLIED ) assert present_effect.already_present_at_first_apply is True present_audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.apply_present") .order_by(AuditLog.created_at.desc()), ) assert present_audit is not None assert present_audit.detail["already_applied_count"] == 1 assert ( present_audit.detail["results"][0]["already_present_at_first_apply"] is True ) revoke_result = await DataAccessService().revoke_data_access_binding( db_session, actor=owner, binding_id=present_effect.binding_id, reason="group_owner_revoke", trace_id="dax-035-revoke-adopted-test", ) assert revoke_result.revoked_pending_count == 1 assert revoke_result.status == BINDING_STATUS_REVOKED_PENDING absent_effect = await db_session.scalar( select(DataAccessCompiledEffect) .where(DataAccessCompiledEffect.binding_id == present_effect.binding_id) .where(DataAccessCompiledEffect.desired_state == EFFECT_DESIRED_ABSENT) .order_by(DataAccessCompiledEffect.created_at.desc()), ) assert absent_effect is not None assert absent_effect.provider_member == present_effect.provider_member assert absent_effect.provider_role == present_effect.provider_role assert absent_effect.target_project_id == present_effect.target_project_id assert absent_effect.target_dataset_id == present_effect.target_dataset_id assert absent_effect.already_present_at_first_apply is False absent_driver = FakeApplyDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-adopt-revoke-1", bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=( "group:adopted-analytics@example.com", "group:unrelated-analytics@example.com", ), ), ), raw_policy={"etag": "provider-etag-adopt-revoke-1"}, ), }, apply_results=[ DataAccessDatasetIamUpdateResult( ok=True, changed=True, etag="provider-etag-adopt-revoke-2", ), ], ) absent_result = await apply_absent_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=absent_driver, trace_id="dax-035-apply-adopted-revoke-test", ) assert len(absent_driver.set_calls) == 1 project_id, dataset_id, expected_etag, absent_patch = absent_driver.set_calls[0] assert (project_id, dataset_id, expected_etag) == ( "helix-data-prod", "finance", "provider-etag-adopt-revoke-1", ) assert absent_patch == DataAccessDatasetIamPolicyPatch( add_bindings=(), remove_bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=("group:adopted-analytics@example.com",), ), ), ) assert absent_result.removed_count == 1 assert absent_result.results[0].effect_id == absent_effect.id assert absent_effect.status == EFFECT_STATUS_APPLIED assert absent_effect.last_provider_etag == "provider-etag-adopt-revoke-2" binding = await db_session.get(DataAccessBinding, present_effect.binding_id) assert binding is not None assert binding.status == BINDING_STATUS_REVOKED async def test_apply_present_ignores_absent_effects_for_revoke_slice( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-absent-ignore", config_public={ "apply_mode_enabled": True, }, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("absent-ignored@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, desired_state=EFFECT_DESIRED_ABSENT, binding_status=BINDING_STATUS_REVOKED_PENDING, ) driver = FakeApplyDriver(error=AssertionError("provider must not be called")) result = await apply_present_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-033-absent-ignored-test", ) assert driver.get_calls == [] assert driver.set_calls == [] assert result.selected_effect_count == 1 assert result.selected_present_effect_count == 0 assert result.ignored_non_present_effect_count == 1 assert result.allowed_effect_count == 0 assert result.blocked_effect_count == 0 assert result.gate_decisions == () assert result.results == () assert effect.status == EFFECT_STATUS_PENDING_RECONCILE assert effect.last_provider_outcome is None async def test_apply_present_conflict_marks_effect_retryable( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-conflict", config_public={ "apply_mode_enabled": True, }, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("conflict-analytics@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() request = _request( requester_id=owner.id, resource=resource, group=group, entitlement=entitlement, ) db_session.add(request) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, request=request, item=request.items[0], ) driver = FakeApplyDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-conflict-local", bindings=(), raw_policy={"etag": "provider-etag-conflict-local"}, ), }, apply_results=[ DataAccessDatasetIamUpdateResult( ok=False, changed=False, etag="provider-etag-conflict-remote", conflict=True, error_message="etag changed before apply", ), ], ) result = await apply_present_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-033-conflict-test", ) assert driver.get_calls == [("helix-data-prod", "finance")] assert len(driver.set_calls) == 1 assert result.provider_write_count == 1 assert result.applied_count == 0 assert result.conflict_count == 1 assert result.error_count == 0 assert result.results[0].outcome == APPLY_PRESENT_OUTCOME_CONFLICT assert effect.status == EFFECT_STATUS_RETRYABLE assert effect.last_provider_outcome == APPLY_PRESENT_OUTCOME_CONFLICT assert effect.last_provider_etag == "provider-etag-conflict-remote" assert effect.last_error == "etag changed before apply" binding = await db_session.get(DataAccessBinding, effect.binding_id) assert binding is not None assert binding.reconcile_status == RECONCILE_STATUS_RETRYABLE assert binding.last_reconcile_error == "etag changed before apply" assert request.reconcile_status == RECONCILE_STATUS_RETRYABLE assert request.last_reconcile_error == "etag changed before apply" audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.apply_present") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.status == "failure" assert audit.detail["conflict_count"] == 1 assert audit.detail["results"][0]["outcome"] == APPLY_PRESENT_OUTCOME_CONFLICT async def test_apply_present_auth_failure_marks_effect_retryable_without_apply( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-auth-failure", config_public={ "apply_mode_enabled": True, }, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("auth-failure-analytics@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() request = _request( requester_id=owner.id, resource=resource, group=group, entitlement=entitlement, ) db_session.add(request) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, request=request, item=request.items[0], ) driver = FakeApplyDriver( error=RuntimeError( "provider auth failed -----BEGIN PRIVATE KEY-----\n" "do-not-store-me\n-----END PRIVATE KEY-----", ), ) result = await apply_present_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-036-present-auth-failure-test", ) assert driver.get_calls == [("helix-data-prod", "finance")] assert driver.set_calls == [] assert result.provider_write_count == 0 assert result.applied_count == 0 assert result.already_applied_count == 0 assert result.error_count == 1 assert result.results[0].outcome == APPLY_PRESENT_OUTCOME_ERROR assert result.results[0].provider_etag is None assert ( result.results[0].error_message == "provider auth failed " ) assert effect.status == EFFECT_STATUS_RETRYABLE assert effect.last_provider_outcome == APPLY_PRESENT_OUTCOME_ERROR assert effect.last_provider_etag is None assert effect.last_error == "provider auth failed " assert effect.already_present_at_first_apply is False binding = await db_session.get(DataAccessBinding, effect.binding_id) assert binding is not None assert binding.status == BINDING_STATUS_ACTIVE assert binding.reconcile_status == RECONCILE_STATUS_RETRYABLE assert binding.last_reconcile_error == ( "provider auth failed " ) assert request.status == REQUEST_STATUS_APPROVED assert request.reconcile_status == RECONCILE_STATUS_RETRYABLE assert request.last_reconcile_error == ( "provider auth failed " ) audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.apply_present") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.status == "failure" assert audit.detail["provider_mutation_attempted"] is False assert audit.detail["provider_write_count"] == 0 assert audit.detail["error_count"] == 1 assert audit.detail["results"][0]["outcome"] == APPLY_PRESENT_OUTCOME_ERROR assert ( audit.detail["results"][0]["error_message"] == "provider auth failed " ) assert "do-not-store-me" not in audit.detail["results"][0]["error_message"] async def test_apply_absent_canary_effect_removes_exact_member_and_marks_revoked( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-absent", config_public={ "apply_mode_enabled": True, }, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("revoke-analytics@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() request = _request( requester_id=owner.id, resource=resource, group=group, entitlement=entitlement, ) db_session.add(request) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, request=request, item=request.items[0], desired_state=EFFECT_DESIRED_ABSENT, binding_status=BINDING_STATUS_REVOKED_PENDING, ) driver = FakeApplyDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-revoke-1", bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=( "group:revoke-analytics@example.com", "group:unrelated@example.com", ), ), DataAccessDatasetIamBinding( role="roles/bigquery.metadataViewer", members=("group:metadata-only@example.com",), ), ), raw_policy={"etag": "provider-etag-revoke-1"}, ), }, apply_results=[ DataAccessDatasetIamUpdateResult( ok=True, changed=True, etag="provider-etag-revoke-2", ), ], ) result = await apply_absent_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-034-apply-absent-test", ) assert driver.get_calls == [("helix-data-prod", "finance")] assert len(driver.set_calls) == 1 project_id, dataset_id, expected_etag, patch = driver.set_calls[0] assert (project_id, dataset_id, expected_etag) == ( "helix-data-prod", "finance", "provider-etag-revoke-1", ) assert patch == DataAccessDatasetIamPolicyPatch( add_bindings=(), remove_bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=("group:revoke-analytics@example.com",), ), ), ) assert result.selected_effect_count == 1 assert result.selected_absent_effect_count == 1 assert result.ignored_non_absent_effect_count == 0 assert result.allowed_effect_count == 1 assert result.blocked_effect_count == 0 assert result.provider_write_count == 1 assert result.removed_count == 1 assert result.already_absent_count == 0 assert result.conflict_count == 0 assert result.error_count == 0 assert result.results[0].effect_id == effect.id assert result.results[0].outcome == APPLY_ABSENT_OUTCOME_APPLIED assert effect.status == EFFECT_STATUS_APPLIED assert effect.last_provider_outcome == APPLY_ABSENT_OUTCOME_APPLIED assert effect.last_provider_etag == "provider-etag-revoke-2" assert effect.last_error is None binding = await db_session.get(DataAccessBinding, effect.binding_id) assert binding is not None assert binding.status == BINDING_STATUS_REVOKED assert binding.reconcile_status == RECONCILE_STATUS_APPLIED assert binding.last_reconcile_run_id == result.run_id assert request.reconcile_status == RECONCILE_STATUS_APPLIED assert request.last_reconcile_run_id == result.run_id run = await db_session.scalar( select(DataAccessEffectRun).where(DataAccessEffectRun.effect_id == effect.id), ) assert run is not None assert run.run_id == result.run_id assert run.mode == "apply_absent" assert run.request_id == request.id assert run.effect_status == EFFECT_STATUS_APPLIED assert run.provider_outcome == APPLY_ABSENT_OUTCOME_APPLIED assert run.provider_etag == "provider-etag-revoke-2" audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.apply_absent") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.status == "success" assert audit.detail["provider_mutation_attempted"] is True assert audit.detail["provider_write_count"] == 1 assert audit.detail["removed_count"] == 1 assert audit.detail["already_absent_count"] == 0 assert audit.detail["gate_decisions"] == [{ "effect_id": str(effect.id), "resource_id": str(resource.id), "target_project_id": "helix-data-prod", "target_dataset_id": "finance", "provider_member": "group:revoke-analytics@example.com", "provider_role": c.BIGQUERY_DATA_VIEWER_ROLE, "desired_state": EFFECT_DESIRED_ABSENT, "locator": "helix-data-prod.finance", "allowed": True, "reason": APPLY_GATE_REASON_ALLOWED, }] assert audit.detail["results"][0]["outcome"] == APPLY_ABSENT_OUTCOME_APPLIED async def test_apply_absent_already_absent_is_idempotent( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-already-absent", config_public={ "apply_mode_enabled": True, }, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("already-absent@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, desired_state=EFFECT_DESIRED_ABSENT, binding_status=BINDING_STATUS_REVOKED_PENDING, ) driver = FakeApplyDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-already-absent", bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=("group:unrelated@example.com",), ), ), raw_policy={"etag": "provider-etag-already-absent"}, ), }, apply_results=[ DataAccessDatasetIamUpdateResult( ok=True, changed=False, etag="provider-etag-already-absent", already_applied=True, ), ], ) result = await apply_absent_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-034-already-absent-test", ) assert result.removed_count == 0 assert result.already_absent_count == 1 assert result.error_count == 0 assert result.results[0].outcome == APPLY_ABSENT_OUTCOME_ALREADY_ABSENT assert effect.status == EFFECT_STATUS_APPLIED assert effect.last_provider_outcome == APPLY_ABSENT_OUTCOME_ALREADY_ABSENT assert effect.last_provider_etag == "provider-etag-already-absent" binding = await db_session.get(DataAccessBinding, effect.binding_id) assert binding is not None assert binding.status == BINDING_STATUS_REVOKED async def test_apply_absent_ignores_present_effects_for_grant_slice( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-present-ignore", config_public={ "apply_mode_enabled": True, }, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("present-ignored@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, desired_state=EFFECT_DESIRED_PRESENT, ) driver = FakeApplyDriver(error=AssertionError("provider must not be called")) result = await apply_absent_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-034-present-ignored-test", ) assert driver.get_calls == [] assert driver.set_calls == [] assert result.selected_effect_count == 1 assert result.selected_absent_effect_count == 0 assert result.ignored_non_absent_effect_count == 1 assert result.allowed_effect_count == 0 assert result.blocked_effect_count == 0 assert result.gate_decisions == () assert result.results == () assert effect.status == EFFECT_STATUS_PENDING_RECONCILE assert effect.last_provider_outcome is None async def test_apply_absent_ignores_absent_effect_for_active_binding( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-absent-active-ignore", config_public={ "apply_mode_enabled": True, }, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("active-absent-ignored@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, desired_state=EFFECT_DESIRED_ABSENT, binding_status=BINDING_STATUS_ACTIVE, ) driver = FakeApplyDriver(error=AssertionError("provider must not be called")) result = await apply_absent_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-034-active-absent-ignored-test", ) assert driver.get_calls == [] assert driver.set_calls == [] assert result.selected_effect_count == 1 assert result.selected_absent_effect_count == 0 assert result.ignored_non_absent_effect_count == 0 assert result.ignored_non_revoked_binding_effect_count == 1 assert result.allowed_effect_count == 0 assert result.blocked_effect_count == 0 assert result.gate_decisions == () assert result.results == () assert effect.status == EFFECT_STATUS_PENDING_RECONCILE assert effect.last_provider_outcome is None binding = await db_session.get(DataAccessBinding, effect.binding_id) assert binding is not None assert binding.status == BINDING_STATUS_ACTIVE async def test_apply_absent_conflict_marks_effect_retryable_without_revoking( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-absent-conflict", config_public={ "apply_mode_enabled": True, }, ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("conflict-revoke@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() request = _request( requester_id=owner.id, resource=resource, group=group, entitlement=entitlement, ) db_session.add(request) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, request=request, item=request.items[0], desired_state=EFFECT_DESIRED_ABSENT, binding_status=BINDING_STATUS_REVOKED_PENDING, ) driver = FakeApplyDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-revoke-conflict-local", bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=("group:conflict-revoke@example.com",), ), ), raw_policy={"etag": "provider-etag-revoke-conflict-local"}, ), }, apply_results=[ DataAccessDatasetIamUpdateResult( ok=False, changed=False, etag="provider-etag-revoke-conflict-remote", conflict=True, error_message="etag changed before revoke apply", ), ], ) result = await apply_absent_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-034-absent-conflict-test", ) assert driver.get_calls == [("helix-data-prod", "finance")] assert len(driver.set_calls) == 1 assert result.provider_write_count == 1 assert result.removed_count == 0 assert result.conflict_count == 1 assert result.error_count == 0 assert result.results[0].outcome == APPLY_ABSENT_OUTCOME_CONFLICT assert effect.status == EFFECT_STATUS_RETRYABLE assert effect.last_provider_outcome == APPLY_ABSENT_OUTCOME_CONFLICT assert effect.last_provider_etag == "provider-etag-revoke-conflict-remote" assert effect.last_error == "etag changed before revoke apply" binding = await db_session.get(DataAccessBinding, effect.binding_id) assert binding is not None assert binding.status == BINDING_STATUS_REVOKED_PENDING assert binding.reconcile_status == RECONCILE_STATUS_RETRYABLE assert binding.last_reconcile_error == "etag changed before revoke apply" assert request.reconcile_status == RECONCILE_STATUS_RETRYABLE assert request.last_reconcile_error == "etag changed before revoke apply" audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.apply_absent") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.status == "failure" assert audit.detail["conflict_count"] == 1 assert audit.detail["results"][0]["outcome"] == APPLY_ABSENT_OUTCOME_CONFLICT async def test_apply_absent_partial_failure_does_not_mass_revoke( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-apply-absent-partial-failure", config_public={ "apply_mode_enabled": True, }, ) entitlement = _entitlement() finance_resource = _resource( integration_id=integration_id, dataset_id="finance", owner_person_id=owner.id, ) hr_resource = _resource( integration_id=integration_id, dataset_id="hr", owner_person_id=owner.id, ) finance_group = _group("finance-revoke@example.com") hr_group = _group("hr-revoke@example.com") db_session.add_all([ entitlement, finance_resource, hr_resource, finance_group, hr_group, ]) await db_session.flush() finance_request = _request( requester_id=owner.id, resource=finance_resource, group=finance_group, entitlement=entitlement, ) hr_request = _request( requester_id=owner.id, resource=hr_resource, group=hr_group, entitlement=entitlement, ) db_session.add_all([finance_request, hr_request]) await db_session.flush() finance_effect = await _seed_effect( db_session, resource=finance_resource, group=finance_group, entitlement=entitlement, request=finance_request, item=finance_request.items[0], desired_state=EFFECT_DESIRED_ABSENT, binding_status=BINDING_STATUS_REVOKED_PENDING, ) hr_effect = await _seed_effect( db_session, resource=hr_resource, group=hr_group, entitlement=entitlement, request=hr_request, item=hr_request.items[0], desired_state=EFFECT_DESIRED_ABSENT, binding_status=BINDING_STATUS_REVOKED_PENDING, ) driver = FakeApplyDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-finance-partial-1", bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=("group:finance-revoke@example.com",), ), ), raw_policy={"etag": "provider-etag-finance-partial-1"}, ), ("helix-data-prod", "hr"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="hr", etag="provider-etag-hr-partial-1", bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=("group:hr-revoke@example.com",), ), ), raw_policy={"etag": "provider-etag-hr-partial-1"}, ), }, apply_results=[ DataAccessDatasetIamUpdateResult( ok=True, changed=True, etag="provider-etag-finance-partial-2", ), DataAccessDatasetIamUpdateResult( ok=False, changed=False, error_message="bigquery auth failed for hr dataset", ), ], ) result = await apply_absent_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-036-absent-partial-failure-test", ) assert driver.get_calls == [ ("helix-data-prod", "finance"), ("helix-data-prod", "hr"), ] assert len(driver.set_calls) == 2 assert result.provider_write_count == 2 assert result.removed_count == 1 assert result.error_count == 1 results_by_effect_id = {row.effect_id: row for row in result.results} assert ( results_by_effect_id[finance_effect.id].outcome == APPLY_ABSENT_OUTCOME_APPLIED ) assert results_by_effect_id[finance_effect.id].provider_etag == ( "provider-etag-finance-partial-2" ) assert ( results_by_effect_id[hr_effect.id].outcome == APPLY_ABSENT_OUTCOME_ERROR ) assert results_by_effect_id[hr_effect.id].provider_etag == ( "provider-etag-hr-partial-1" ) assert results_by_effect_id[hr_effect.id].error_message == ( "bigquery auth failed for hr dataset" ) finance_binding = await db_session.get( DataAccessBinding, finance_effect.binding_id, ) hr_binding = await db_session.get(DataAccessBinding, hr_effect.binding_id) assert finance_binding is not None assert hr_binding is not None assert finance_effect.status == EFFECT_STATUS_APPLIED assert finance_binding.status == BINDING_STATUS_REVOKED assert finance_binding.reconcile_status == RECONCILE_STATUS_APPLIED assert finance_request.reconcile_status == RECONCILE_STATUS_APPLIED assert hr_effect.status == EFFECT_STATUS_RETRYABLE assert hr_effect.last_provider_outcome == APPLY_ABSENT_OUTCOME_ERROR assert hr_effect.last_error == "bigquery auth failed for hr dataset" assert hr_binding.status == BINDING_STATUS_REVOKED_PENDING assert hr_binding.reconcile_status == RECONCILE_STATUS_RETRYABLE assert hr_binding.last_reconcile_error == "bigquery auth failed for hr dataset" assert hr_request.reconcile_status == RECONCILE_STATUS_RETRYABLE assert hr_request.last_reconcile_error == "bigquery auth failed for hr dataset" audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.apply_absent") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.status == "failure" assert audit.detail["removed_count"] == 1 assert audit.detail["error_count"] == 1 audit_results_by_effect_id = { uuid.UUID(row["effect_id"]): row for row in audit.detail["results"] } assert audit_results_by_effect_id[finance_effect.id]["outcome"] == ( APPLY_ABSENT_OUTCOME_APPLIED ) assert audit_results_by_effect_id[hr_effect.id]["outcome"] == ( APPLY_ABSENT_OUTCOME_ERROR ) async def test_observe_reconcile_records_deltas_without_apply( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id(db_session) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) groups = [ _group("existing@example.com"), _group("missing@example.com"), _group("remove@example.com"), _group("already-absent@example.com"), ] db_session.add_all([entitlement, resource, *groups]) await db_session.flush() existing = await _seed_effect( db_session, resource=resource, group=groups[0], entitlement=entitlement, desired_state=EFFECT_DESIRED_PRESENT, ) missing = await _seed_effect( db_session, resource=resource, group=groups[1], entitlement=entitlement, desired_state=EFFECT_DESIRED_PRESENT, ) remove = await _seed_effect( db_session, resource=resource, group=groups[2], entitlement=entitlement, desired_state=EFFECT_DESIRED_ABSENT, binding_status=BINDING_STATUS_REVOKED_PENDING, ) already_absent = await _seed_effect( db_session, resource=resource, group=groups[3], entitlement=entitlement, desired_state=EFFECT_DESIRED_ABSENT, binding_status=BINDING_STATUS_REVOKED_PENDING, ) driver = FakeObserveDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-1", bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=( "group:existing@example.com", "group:remove@example.com", ), ), ), raw_policy={"etag": "provider-etag-1"}, ), }, ) result = await observe_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-024-observe-test", ) assert driver.get_calls == [("helix-data-prod", "finance")] assert driver.set_calls == [] assert driver.forbidden_calls == [] assert result.observed_effect_count == 4 assert result.would_add_count == 1 assert result.would_remove_count == 1 assert result.unchanged_count == 2 assert result.error_count == 0 assert result.unmanaged_grant_count == 0 outcomes = { row.id: row.last_provider_outcome for row in ( await db_session.scalars(select(DataAccessCompiledEffect)) ).all() } assert outcomes == { existing.id: OBSERVE_OUTCOME_UNCHANGED, missing.id: OBSERVE_OUTCOME_WOULD_ADD, remove.id: OBSERVE_OUTCOME_WOULD_REMOVE, already_absent.id: OBSERVE_OUTCOME_UNCHANGED, } for effect in (existing, missing, remove, already_absent): assert effect.status == EFFECT_STATUS_OBSERVED assert effect.last_provider_etag == "provider-etag-1" assert effect.last_error is None audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.observe") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.status == "success" assert audit.detail["mode"] == "observe" assert audit.detail["provider_mutation_enabled"] is False assert audit.detail["would_add_count"] == 1 assert audit.detail["would_remove_count"] == 1 assert audit.detail["unchanged_count"] == 2 assert audit.detail["error_count"] == 0 assert audit.detail["unmanaged_grant_count"] == 0 assert audit.detail["run_id"] == str(result.run_id) assert set(audit.detail).issuperset({ "would_add_count", "would_remove_count", "unchanged_count", "unmanaged_grant_count", "error_count", }) assert {delta["effect_id"] for delta in audit.detail["deltas"]} == { str(existing.id), str(missing.id), str(remove.id), str(already_absent.id), } runs = ( await db_session.scalars( select(DataAccessEffectRun) .where(DataAccessEffectRun.run_id == result.run_id) .order_by(DataAccessEffectRun.provider_member), ) ).all() assert len(runs) == 4 assert {run.effect_id for run in runs} == { existing.id, missing.id, remove.id, already_absent.id, } assert {run.effect_status for run in runs} == {EFFECT_STATUS_OBSERVED} assert {run.provider_outcome for run in runs} == { OBSERVE_OUTCOME_UNCHANGED, OBSERVE_OUTCOME_WOULD_ADD, OBSERVE_OUTCOME_WOULD_REMOVE, } async def test_observe_reconcile_is_bounded_to_selected_integration_and_effects( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id(db_session, slug_prefix="dax-observe-a") other_integration_id = await _integration_id( db_session, slug_prefix="dax-observe-b", ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, dataset_id="finance", owner_person_id=owner.id, ) ignored_resource = _resource( integration_id=other_integration_id, dataset_id="hr", owner_person_id=owner.id, ) group = _group("analytics@example.com") ignored_group = _group("hr@example.com") db_session.add_all([entitlement, resource, ignored_resource, group, ignored_group]) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, ) ignored_effect = await _seed_effect( db_session, resource=ignored_resource, group=ignored_group, entitlement=entitlement, ) driver = FakeObserveDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-2", bindings=(), raw_policy={"etag": "provider-etag-2"}, ), }, ) result = await observe_reconcile_compiled_effects( db_session, integration_id=integration_id, effect_ids=[effect.id, ignored_effect.id], driver=driver, trace_id="dax-024-bounded-test", ) assert driver.get_calls == [("helix-data-prod", "finance")] assert driver.forbidden_calls == [] assert result.observed_effect_count == 1 assert result.would_add_count == 1 assert effect.status == EFFECT_STATUS_OBSERVED assert ignored_effect.status == EFFECT_STATUS_PENDING_RECONCILE assert ignored_effect.last_provider_outcome is None async def test_observe_reconcile_uses_locator_reads_and_explicit_count_report( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-negative-guard", ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, dataset_id="finance", owner_person_id=owner.id, ) ignored_resource = _resource( integration_id=integration_id, dataset_id="hr", owner_person_id=owner.id, ) group = _group("selected-dax@example.com") ignored_group = _group("ignored-dax@example.com") db_session.add_all([entitlement, resource, ignored_resource, group, ignored_group]) await db_session.flush() selected_effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, ) ignored_effect = await _seed_effect( db_session, resource=ignored_resource, group=ignored_group, entitlement=entitlement, ) driver = FakeObserveDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-negative-guard", bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=("group:unmanaged-negative@example.com",), ), ), raw_policy={"etag": "provider-etag-negative-guard"}, ), }, ) result = await observe_reconcile_compiled_effects( db_session, integration_id=integration_id, effect_ids=[selected_effect.id], resource_ids=[resource.id], driver=driver, trace_id="dax-028-negative-iam-guard-test", ) assert driver.get_calls == [("helix-data-prod", "finance")] assert driver.set_calls == [] assert driver.forbidden_calls == [] assert result.observed_effect_count == 1 assert result.would_add_count == 1 assert result.would_remove_count == 0 assert result.unchanged_count == 0 assert result.unmanaged_grant_count == 1 assert result.error_count == 0 assert selected_effect.status == EFFECT_STATUS_OBSERVED assert ignored_effect.status == EFFECT_STATUS_PENDING_RECONCILE assert ignored_effect.last_provider_outcome is None audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.observe") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.detail["selection"] == { "effect_ids": [str(selected_effect.id)], "resource_ids": [str(resource.id)], } assert audit.detail["would_add_count"] == 1 assert audit.detail["would_remove_count"] == 0 assert audit.detail["unchanged_count"] == 0 assert audit.detail["unmanaged_grant_count"] == 1 assert audit.detail["error_count"] == 0 assert audit.detail["deltas"] == [{ "effect_id": str(selected_effect.id), "resource_id": str(resource.id), "target_project_id": "helix-data-prod", "target_dataset_id": "finance", "provider_member": "group:selected-dax@example.com", "provider_role": c.BIGQUERY_DATA_VIEWER_ROLE, "desired_state": EFFECT_DESIRED_PRESENT, "outcome": OBSERVE_OUTCOME_WOULD_ADD, "provider_etag": "provider-etag-negative-guard", "error_message": None, }] assert audit.detail["unmanaged_grants"] == [{ "resource_id": str(resource.id), "target_project_id": "helix-data-prod", "target_dataset_id": "finance", "provider_member": "group:unmanaged-negative@example.com", "provider_role": c.BIGQUERY_DATA_VIEWER_ROLE, "provider_etag": "provider-etag-negative-guard", }] async def test_observe_reconcile_can_be_bounded_to_resource_ids( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id(db_session, slug_prefix="dax-observe-r") entitlement = _entitlement() finance_resource = _resource( integration_id=integration_id, dataset_id="finance", owner_person_id=owner.id, ) hr_resource = _resource( integration_id=integration_id, dataset_id="hr", owner_person_id=owner.id, ) finance_group = _group("finance@example.com") hr_group = _group("hr@example.com") db_session.add_all([ entitlement, finance_resource, hr_resource, finance_group, hr_group, ]) await db_session.flush() finance_effect = await _seed_effect( db_session, resource=finance_resource, group=finance_group, entitlement=entitlement, ) hr_effect = await _seed_effect( db_session, resource=hr_resource, group=hr_group, entitlement=entitlement, ) driver = FakeObserveDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-finance", bindings=(), raw_policy={"etag": "provider-etag-finance"}, ), }, ) result = await observe_reconcile_compiled_effects( db_session, integration_id=integration_id, resource_ids=[finance_resource.id], driver=driver, trace_id="dax-025-resource-scope-test", ) assert driver.get_calls == [("helix-data-prod", "finance")] assert driver.forbidden_calls == [] assert result.observed_effect_count == 1 assert result.would_add_count == 1 assert finance_effect.status == EFFECT_STATUS_OBSERVED assert hr_effect.status == EFFECT_STATUS_PENDING_RECONCILE assert hr_effect.last_provider_outcome is None audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.observe") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.detail["lock_strategy"] == "for_update_skip_locked" assert audit.detail["selection"]["resource_ids"] == [str(finance_resource.id)] async def test_observe_reconcile_reports_unmanaged_grants_without_adopting( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id(db_session, slug_prefix="dax-unmanaged") entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) observed_group = _group("already-managed@example.com") pending_group = _group("pending-managed@example.com") db_session.add_all([entitlement, resource, observed_group, pending_group]) await db_session.flush() observed_effect = await _seed_effect( db_session, resource=resource, group=observed_group, entitlement=entitlement, ) observed_effect.status = EFFECT_STATUS_OBSERVED pending_effect = await _seed_effect( db_session, resource=resource, group=pending_group, entitlement=entitlement, ) before_bindings = ( await db_session.scalars(select(DataAccessBinding)) ).all() driver = FakeObserveDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-unmanaged", bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=( "group:already-managed@example.com", "group:pending-managed@example.com", "group:unmanaged@example.com", "user:ignored@example.com", ), ), DataAccessDatasetIamBinding( role="roles/bigquery.metadataViewer", members=("group:metadata-only@example.com",), ), ), raw_policy={"etag": "provider-etag-unmanaged"}, ), }, ) result = await observe_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-027-unmanaged-report-test", ) after_bindings = ( await db_session.scalars(select(DataAccessBinding)) ).all() assert driver.forbidden_calls == [] assert len(after_bindings) == len(before_bindings) assert pending_effect.status == EFFECT_STATUS_OBSERVED assert observed_effect.status == EFFECT_STATUS_OBSERVED assert result.observed_effect_count == 1 assert result.unmanaged_grant_count == 2 unmanaged_by_member = { grant.provider_member: grant for grant in result.unmanaged_grants } unmanaged = unmanaged_by_member["group:unmanaged@example.com"] assert unmanaged.resource_id == resource.id assert unmanaged.target_project_id == "helix-data-prod" assert unmanaged.target_dataset_id == "finance" assert unmanaged.provider_role == c.BIGQUERY_DATA_VIEWER_ROLE assert unmanaged.provider_member == "group:unmanaged@example.com" assert unmanaged.provider_etag == "provider-etag-unmanaged" audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.observe") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.status == "success" assert audit.detail["unmanaged_grant_count"] == 2 assert { grant["provider_member"] for grant in audit.detail["unmanaged_grants"] } == {"group:unmanaged@example.com", "user:ignored@example.com"} async def test_observe_reconcile_unmanaged_report_ignores_non_federated_resources( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id( db_session, slug_prefix="dax-unmanaged-scope", ) entitlement = _entitlement() resource = _resource( integration_id=integration_id, dataset_id="finance", owner_person_id=owner.id, ) unsupported_resource = _resource( integration_id=integration_id, dataset_id="finance_table", resource_type="project", owner_person_id=owner.id, ) group = _group("finance-scope@example.com") table_group = _group("table-scope@example.com") db_session.add_all([entitlement, resource, unsupported_resource, group, table_group]) await db_session.flush() selected_effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, ) ignored_effect = await _seed_effect( db_session, resource=unsupported_resource, group=table_group, entitlement=entitlement, ) driver = FakeObserveDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-federated-only", bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=( "group:finance-scope@example.com", "group:unmanaged-federated@example.com", ), ), ), raw_policy={"etag": "provider-etag-federated-only"}, ), }, ) result = await observe_reconcile_compiled_effects( db_session, integration_id=integration_id, resource_ids=[resource.id, unsupported_resource.id], driver=driver, trace_id="dax-027-non-federated-ignore-test", ) assert driver.get_calls == [("helix-data-prod", "finance")] assert driver.forbidden_calls == [] assert selected_effect.status == EFFECT_STATUS_OBSERVED assert ignored_effect.status == EFFECT_STATUS_PENDING_RECONCILE assert result.observed_effect_count == 1 assert result.unmanaged_grant_count == 1 assert result.unmanaged_grants[0].provider_member == ( "group:unmanaged-federated@example.com" ) async def test_observe_reconcile_rolls_up_success_to_binding_and_request( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id(db_session, slug_prefix="dax-observe-roll") entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("analytics-rollup@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() request = _request( requester_id=owner.id, resource=resource, group=group, entitlement=entitlement, ) db_session.add(request) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, request=request, item=request.items[0], ) driver = FakeObserveDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-rollup", bindings=(), raw_policy={"etag": "provider-etag-rollup"}, ), }, ) result = await observe_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-026-rollup-success-test", ) binding = await db_session.get(DataAccessBinding, effect.binding_id) assert binding is not None assert binding.reconcile_status == RECONCILE_STATUS_OBSERVED assert binding.last_reconcile_run_id == result.run_id assert binding.last_reconcile_error is None assert request.reconcile_status == RECONCILE_STATUS_OBSERVED assert request.last_reconcile_run_id == result.run_id assert request.last_reconcile_error is None run = await db_session.scalar( select(DataAccessEffectRun).where(DataAccessEffectRun.effect_id == effect.id), ) assert run is not None assert run.provider_outcome == OBSERVE_OUTCOME_WOULD_ADD assert run.effect_status == EFFECT_STATUS_OBSERVED async def test_observe_reconcile_marks_effects_retryable_on_read_error( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id(db_session) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("analytics@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() request = _request( requester_id=owner.id, resource=resource, group=group, entitlement=entitlement, ) db_session.add(request) await db_session.flush() item = request.items[0] effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, request=request, item=item, ) principal = DataAccessPrincipal( principal_type=c.PRINCIPAL_TYPE_GOOGLE_USER, external_subject_id=str(owner.id), email_snapshot=owner.work_email, status="active", ) db_session.add(principal) await db_session.flush() binding = await db_session.get(DataAccessBinding, effect.binding_id) assert binding is not None request.principal_id = principal.id request.principal_email_group_id = None request.principal_type_snapshot = c.PRINCIPAL_TYPE_GOOGLE_USER request.principal_subject_snapshot = str(owner.id) request.principal_email_snapshot = owner.work_email binding.principal_id = principal.id binding.principal_email_group_id = None binding.principal_type_snapshot = c.PRINCIPAL_TYPE_GOOGLE_USER binding.principal_subject_snapshot = str(owner.id) binding.principal_email_snapshot = owner.work_email effect.principal_id = principal.id effect.principal_email_group_id = None effect.principal_type_snapshot = c.PRINCIPAL_TYPE_GOOGLE_USER effect.principal_subject_snapshot = str(owner.id) effect.provider_member = f"{c.BIGQUERY_USER_MEMBER_PREFIX}{owner.work_email}" effect.provider_access_kind = "userByEmail" await db_session.flush() driver = FakeObserveDriver( error=RuntimeError( "provider auth failed -----BEGIN PRIVATE KEY-----\n" "do-not-store-me\n" "-----END PRIVATE KEY-----", ), ) result = await observe_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-024-error-test", ) assert driver.get_calls == [("helix-data-prod", "finance")] assert driver.set_calls == [] assert result.observed_effect_count == 1 assert result.error_count == 1 assert result.deltas[0].outcome == OBSERVE_OUTCOME_ERROR assert effect.status == EFFECT_STATUS_RETRYABLE assert effect.last_provider_outcome == OBSERVE_OUTCOME_ERROR assert effect.last_provider_etag is None assert effect.last_error == "provider auth failed " assert "do-not-store-me" not in effect.last_error run = await db_session.scalar( select(DataAccessEffectRun).where(DataAccessEffectRun.effect_id == effect.id), ) assert run is not None assert run.run_id == result.run_id assert run.request_id == request.id assert run.binding_id == effect.binding_id assert run.effect_status == EFFECT_STATUS_RETRYABLE assert run.provider_outcome == OBSERVE_OUTCOME_ERROR assert run.error_message == "provider auth failed " assert run.principal_id == principal.id assert run.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER assert run.principal_subject_snapshot == str(owner.id) assert run.provider_access_kind == "userByEmail" binding = await db_session.get(DataAccessBinding, effect.binding_id) assert binding is not None assert binding.reconcile_status == RECONCILE_STATUS_RETRYABLE assert binding.last_reconcile_run_id == result.run_id assert ( binding.last_reconcile_error == "provider auth failed " ) assert request.reconcile_status == RECONCILE_STATUS_RETRYABLE assert request.last_reconcile_run_id == result.run_id assert request.last_reconcile_error == "provider auth failed " audit = await db_session.scalar( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.observe") .order_by(AuditLog.created_at.desc()), ) assert audit is not None assert audit.status == "failure" assert audit.detail["error_count"] == 1 assert ( audit.detail["deltas"][0]["error_message"] == "provider auth failed " ) async def test_observe_reconcile_retries_retryable_effects( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id(db_session) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("analytics@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, effect_status=EFFECT_STATUS_RETRYABLE, ) effect.last_provider_outcome = OBSERVE_OUTCOME_ERROR effect.last_error = "previous provider failure" driver = FakeObserveDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-retry", bindings=( DataAccessDatasetIamBinding( role=c.BIGQUERY_DATA_VIEWER_ROLE, members=("group:analytics@example.com",), ), ), raw_policy={"etag": "provider-etag-retry"}, ), }, ) result = await observe_reconcile_compiled_effects( db_session, integration_id=integration_id, driver=driver, trace_id="dax-024-retryable-test", ) assert driver.get_calls == [("helix-data-prod", "finance")] assert driver.set_calls == [] assert result.observed_effect_count == 1 assert result.unchanged_count == 1 assert effect.status == EFFECT_STATUS_OBSERVED assert effect.last_provider_outcome == OBSERVE_OUTCOME_UNCHANGED assert effect.last_provider_etag == "provider-etag-retry" assert effect.last_error is None async def test_observe_reconcile_empty_effect_selection_skips_driver( db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id(db_session) entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("analytics@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, ) driver = FakeObserveDriver(error=AssertionError("driver should not be called")) result = await observe_reconcile_compiled_effects( db_session, integration_id=integration_id, effect_ids=[], driver=driver, trace_id="dax-024-empty-selection-test", ) assert driver.get_calls == [] assert result.observed_effect_count == 0 assert effect.status == EFFECT_STATUS_PENDING_RECONCILE assert effect.last_provider_outcome is None async def test_concurrent_observe_reconcile_skips_locked_effects( db_engine, db_session, employee_user, ) -> None: owner, _ = employee_user integration_id = await _integration_id(db_session, slug_prefix="dax-observe-lock") entitlement = _entitlement() resource = _resource( integration_id=integration_id, owner_person_id=owner.id, ) group = _group("analytics-lock@example.com") db_session.add_all([entitlement, resource, group]) await db_session.flush() effect = await _seed_effect( db_session, resource=resource, group=group, entitlement=entitlement, ) await db_session.commit() first_get_started = asyncio.Event() release_first_get = asyncio.Event() first_driver = FakeObserveDriver( policies={ ("helix-data-prod", "finance"): DataAccessDatasetIamPolicy( project_id="helix-data-prod", dataset_id="finance", etag="provider-etag-lock", bindings=(), raw_policy={"etag": "provider-etag-lock"}, ), }, get_started=first_get_started, wait_before_return=release_first_get, ) second_driver = FakeObserveDriver( error=AssertionError("second worker must not call the provider"), ) session_factory = async_sessionmaker( db_engine, class_=AsyncSession, expire_on_commit=False, ) async def run_first_worker(): async with session_factory() as session: result = await observe_reconcile_compiled_effects( session, integration_id=integration_id, effect_ids=[effect.id], driver=first_driver, trace_id="dax-025-lock-first", ) await session.commit() return result first_task = asyncio.create_task(run_first_worker()) await asyncio.wait_for(first_get_started.wait(), timeout=5) async with session_factory() as second_session: second_result = await observe_reconcile_compiled_effects( second_session, integration_id=integration_id, effect_ids=[effect.id], driver=second_driver, trace_id="dax-025-lock-second", ) await second_session.commit() release_first_get.set() first_result = await asyncio.wait_for(first_task, timeout=5) assert first_driver.get_calls == [("helix-data-prod", "finance")] assert second_driver.get_calls == [] assert first_result.observed_effect_count == 1 assert first_result.would_add_count == 1 assert second_result.observed_effect_count == 0 async with session_factory() as verify_session: stored_effect = await verify_session.get(DataAccessCompiledEffect, effect.id) assert stored_effect is not None assert stored_effect.status == EFFECT_STATUS_OBSERVED assert stored_effect.last_provider_outcome == OBSERVE_OUTCOME_WOULD_ADD audits = ( await verify_session.scalars( select(AuditLog) .where(AuditLog.action == "data_access.reconcile.observe") .where(AuditLog.resource_id == integration_id) ) ).all() assert sorted(audit.detail["observed_effect_count"] for audit in audits) == [0, 1] assert all( audit.detail["selection"]["effect_ids"] == [str(effect.id)] for audit in audits ) def test_observe_reconciler_has_no_apply_or_broad_scan_path() -> None: source = ( Path(__file__).resolve().parents[2] / "app" / "domains" / "data_access" / "reconciler.py" ).read_text(encoding="utf-8") observe_source = source.split( "async def observe_reconcile_compiled_effects", 1, )[1].split("async def resolve_apply_reconcile_gate", 1)[0] assert "set_dataset_iam(" not in observe_source assert source.count("await driver.set_dataset_iam(") == 1 assert source.count("await driver.set_table_iam(") == 1 assert "list_datasets" not in source assert "list_all_datasets" not in source assert "list_resources" not in source assert "list_all_resources" not in source assert "scan_resources" not in source assert ".setIamPolicy" not in source assert ":setIamPolicy" not in source assert "bigquery.jobUser" not in source assert "get_project_iam" not in source assert "set_project_iam" not in source assert "get_project_policy" not in source assert "set_project_policy" not in source assert "grant_access" not in source assert "revoke_access" not in source