from __future__ import annotations import json import uuid from datetime import UTC, datetime from types import SimpleNamespace from typing import Any import httpx import pytest from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.approval.models import ApprovalRequest, ApprovalSubmissionIntent from app.approval.service import ApprovalService 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, EFFECT_DESIRED_PRESENT, EFFECT_MANAGED_REASON_GRANT_REQUEST, EFFECT_STATUS_PENDING_RECONCILE, ENTITLEMENT_STATUS_ACTIVE, ENTITLEMENT_VERSION_V1, OWNER_RESOLUTION_RESOLVED, OWNER_SOURCE_BOOTSTRAP, REQUEST_ITEM_STATUS_APPROVED, REQUEST_ITEM_STATUS_PENDING, REQUEST_ITEM_STATUS_REJECTED, REQUEST_ITEM_STATUS_STALE, REQUEST_STATUS_APPROVED, REQUEST_STATUS_PENDING_FEISHU, REQUEST_STATUS_REJECTED, REQUEST_STATUS_STALE_REQUIRES_RESUBMIT, RESOURCE_STATUS_ACTIVE, DataAccessBinding, DataAccessCompiledEffect, DataAccessEffectRun, DataAccessEntitlement, DataAccessPrincipal, DataAccessRequest, DataAccessRequestItem, DataAccessResource, ) from app.domains.data_access.service import ( DATA_ACCESS_AUDIT_DETAIL_SOFT_LIMIT_BYTES, DATA_ACCESS_COMPILER_VERSION, DATA_ACCESS_FEISHU_PAYLOAD_VERSION, DataAccessService, DataAccessSubmissionError, ) from app.domains.email_group.models import EmailGroup, EmailGroupOwner from app.identity.person_models import IdentityPerson from app.integrations.driver_base import DataAccessDatasetIamPolicy from app.integrations.models import Integration, IntegrationApprovalFlow from app.scheduled_jobs.data_access_handlers import run_data_access_bigquery_apply_reconcile from tests.support.approval_intents import make_submission_intent_retries_due from tests.test_data_access.test_observe_reconciler import FakeApplyDriver class FakeFeishuClient: def __init__(self) -> None: self.resolved_emails: list[str] = [] self.created_instances: list[dict[str, Any]] = [] self.cc_instances: list[dict[str, Any]] = [] self.cancelled_instances: list[dict[str, Any]] = [] async def resolve_user_id_by_email(self, email: str) -> str: self.resolved_emails.append(email) return f"ou_{email.split('@', 1)[0]}" async def create_instance( self, *, approval_code: str, user_id: str | None = None, open_id: str | None = None, form: str, node_approver_user_id_list: list[dict[str, Any]] | None = None, node_approver_open_id_list: list[dict[str, Any]] | None = None, node_cc_user_id_list: list[dict[str, Any]] | None = None, node_cc_open_id_list: list[dict[str, Any]] | None = None, instance_uuid: str | None = None, ) -> dict[str, Any]: self.created_instances.append({ "approval_code": approval_code, "user_id": user_id, "open_id": open_id, "form": form, "node_approver_user_id_list": node_approver_user_id_list, "node_approver_open_id_list": node_approver_open_id_list, "node_cc_user_id_list": node_cc_user_id_list, "node_cc_open_id_list": node_cc_open_id_list, "instance_uuid": instance_uuid, }) return { "instance_code": f"FEISHU-DAX-{len(self.created_instances):03d}", "instance_status": "PENDING", } async def cc_instance( self, *, approval_code: str, instance_code: str, user_id: str | None = None, open_id: str | None = None, cc_user_ids: list[str], comment: str | None = None, ) -> dict[str, Any]: self.cc_instances.append({ "approval_code": approval_code, "instance_code": instance_code, "user_id": user_id, "open_id": open_id, "cc_user_ids": cc_user_ids, "comment": comment, }) return { "ok": True, "instance_code": instance_code, "cc_user_ids": cc_user_ids, } async def cancel_instance( self, *, instance_code: str, user_id: str | None = None, open_id: str | None = None, approval_code: str, ) -> dict[str, Any]: self.cancelled_instances.append({ "instance_code": instance_code, "user_id": user_id, "open_id": open_id, "approval_code": approval_code, }) return {"ok": True} class OpenIdFallbackFeishuClient(FakeFeishuClient): def __init__(self) -> None: super().__init__() self.resolved_open_id_emails: list[str] = [] async def resolve_user_id_by_email(self, email: str) -> str: self.resolved_emails.append(email) raise AppException( code="FEISHU_API_ERROR", message=( "Feishu API error 99991672: Access denied. One of the " "following scopes is required: [contact:user.employee_id:readonly]" ), status_code=502, ) async def resolve_open_id_by_email(self, email: str) -> str: self.resolved_open_id_emails.append(email) return f"ou_{email.split('@', 1)[0]}" class MissingOwnerFeishuClient(OpenIdFallbackFeishuClient): async def resolve_user_id_by_email(self, email: str) -> str: self.resolved_emails.append(email) raise AppException( code="FEISHU_USER_NOT_FOUND", message=f"Feishu user id not found for email {email}", status_code=400, ) async def resolve_open_id_by_email(self, email: str) -> str: self.resolved_open_id_emails.append(email) if email == "employee@test.com": return "ou_employee" raise AppException( code="FEISHU_USER_NOT_FOUND", message=f"Feishu open_id not found for email {email}", status_code=400, ) class ResponseLossFeishuClient(FakeFeishuClient): def __init__(self) -> None: super().__init__() self.instance_code = "FEISHU-DAX-RESPONSE-LOSS" async def create_instance(self, **kwargs: Any) -> dict[str, Any]: self.created_instances.append(dict(kwargs)) if len(self.created_instances) == 1: raise httpx.ReadTimeout("provider accepted; response lost") raise AppException( code="FEISHU_API_ERROR", message="Feishu API error 60012: duplicate uuid", status_code=502, ) async def find_instance_by_uuid( self, *, approval_code: str, instance_uuid: str, start_time: datetime, end_time: datetime, **kwargs: Any, ) -> dict[str, Any]: assert instance_uuid == self.created_instances[0]["instance_uuid"] assert start_time < datetime.now(UTC) < end_time return { "instance_code": self.instance_code, "instance_status": "PENDING", "uuid": instance_uuid, } class RejectingFeishuClient(FakeFeishuClient): async def create_instance(self, **kwargs: Any) -> dict[str, Any]: self.created_instances.append(dict(kwargs)) raise AppException( code="FEISHU_REQUEST_REJECTED", message="provider rejected the approval form", status_code=400, ) class UnknownThenRejectingFeishuClient(FakeFeishuClient): async def create_instance(self, **kwargs: Any) -> dict[str, Any]: self.created_instances.append(dict(kwargs)) if len(self.created_instances) == 1: raise httpx.ReadTimeout("provider result unknown") raise AppException( code="FEISHU_REQUEST_REJECTED", message="provider definitely rejected the retry", status_code=400, ) def _payload_schema( *, data_owner_review_node_id: str | None = None, ) -> dict[str, Any]: schema: dict[str, Any] = { "request_summary": { "id": "widget-request-summary", "type": "textarea", }, "requested_access": { "id": "widget-requested-access", "type": "textarea", }, "dataset_list": { "id": "widget-dataset-list", "type": "textarea", }, "business_reason": { "id": "widget-business-reason", "type": "textarea", }, "detail_url": "widget-detail-url", } if data_owner_review_node_id is not None: schema["approval_process"] = { "approver_nodes": { "data_owner_review": { "id": data_owner_review_node_id, "approver_source": "helix_resource_owner", }, }, } return schema async def _seed_feishu_flow( db_session, *, active: bool = True, flow_key: str = c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, payload_schema: dict[str, Any] | None = None, ) -> IntegrationApprovalFlow: integration = Integration( id=uuid.uuid4(), kind="chat_approval", provider="feishu", name="Feishu Approval", slug=f"feishu-dax-submit-{uuid.uuid4().hex[:8]}", enabled=True, is_primary=True, config_public={"app_id": "cli_data_access_test"}, ) db_session.add(integration) await db_session.flush() flow = IntegrationApprovalFlow( id=uuid.uuid4(), integration_id=integration.id, flow_key=flow_key, label="Data access group grant", external_code="DATA-ACCESS-GROUP-GRANT-CODE", payload_schema=payload_schema if payload_schema is not None else _payload_schema(), active=active, accept_external_origin=False, ) db_session.add(flow) await db_session.flush() return flow async def _seed_entitlement(db_session) -> DataAccessEntitlement: entitlement = 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(entitlement) await db_session.flush() return entitlement async def _seed_bigquery_integration(db_session) -> str: integration = Integration( kind=c.DOMAIN_DATA_ACCESS, provider=c.PROVIDER_BIGQUERY, name="Data Access BigQuery", slug=f"dax-bq-submit-{uuid.uuid4().hex[:8]}", enabled=True, is_primary=False, ) db_session.add(integration) await db_session.flush() return str(integration.id) async def _seed_resource( db_session, *, owner_person_id: uuid.UUID, dataset_id: str = "finance", resource_type: str = c.RESOURCE_TYPE_DATASET, status: str = RESOURCE_STATUS_ACTIVE, notes: str | None = None, ) -> DataAccessResource: resource = DataAccessResource( integration_id=await _seed_bigquery_integration(db_session), provider=c.PROVIDER_BIGQUERY, resource_type=resource_type, project_id="helix-data-prod", dataset_id=dataset_id, display_name=f"helix-data-prod.{dataset_id}", requestable=True, status=status, owner_person_id=str(owner_person_id), owner_resolution_status=OWNER_RESOLUTION_RESOLVED, owner_source=OWNER_SOURCE_BOOTSTRAP, source_version=f"catalog-2026-06-10-{dataset_id}", source_etag=f"etag-{dataset_id}", notes=notes, ) db_session.add(resource) await db_session.flush() return resource async def _seed_group( db_session, *, requester_person_id: uuid.UUID, status: str = "active", add_owner: bool = True, ) -> EmailGroup: group = EmailGroup( email="analytics-readers@example.com", name="Analytics Readers", status=status, google_group_id="google-group-analytics", aliases=["analytics-alias@example.com"], ) db_session.add(group) await db_session.flush() if add_owner: db_session.add( EmailGroupOwner( email_group_id=group.id, person_id=requester_person_id, ), ) await db_session.flush() return group async def _valid_submission_setup(db_session, requester) -> tuple[ EmailGroup, list[DataAccessResource], ]: group = await _seed_group(db_session, requester_person_id=requester.id) resources = [ await _seed_resource( db_session, owner_person_id=requester.id, dataset_id="finance", notes="Finance reporting marts for quarterly close and analytics.", ), await _seed_resource( db_session, owner_person_id=requester.id, dataset_id="people", ), ] await _seed_entitlement(db_session) await _seed_feishu_flow(db_session) return group, resources async def test_submission_creates_request_approval_and_feishu_instance( db_session, employee_user, ) -> None: requester, _ = employee_user group, resources = await _valid_submission_setup(db_session, requester) fake_client = FakeFeishuClient() result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal="analytics-alias@example.com", principal_type=c.PRINCIPAL_TYPE_GOOGLE_GROUP, resource_ids=[resource.id for resource in resources], entitlement=c.ENTITLEMENT_READ, business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-016-idempotency", trace_id="trace-dax-016", feishu_client=fake_client, ) assert result.status == REQUEST_STATUS_PENDING_FEISHU assert result.dataset_count == 2 assert result.idempotency_key == "dax-016-idempotency" assert result.feishu_instance_code == "FEISHU-DAX-001" assert fake_client.resolved_emails == ["employee@test.com"] assert len(fake_client.created_instances) == 1 created = fake_client.created_instances[0] assert created["approval_code"] == "DATA-ACCESS-GROUP-GRANT-CODE" assert created["user_id"] == "ou_employee" assert fake_client.cc_instances == [] submitted_form = json.loads(created["form"]) form_by_id = {item["id"]: item for item in submitted_form} assert set(form_by_id) == { "widget-request-summary", "widget-requested-access", "widget-dataset-list", "widget-business-reason", "widget-detail-url", } assert form_by_id["widget-request-summary"]["value"] == ( "Google Group analytics-readers@example.com is requesting " "read access to 2 BigQuery datasets." ) assert form_by_id["widget-requested-access"]["value"] == ( "Grant BigQuery data viewer to Google Group " "analytics-readers@example.com." ) assert "1. helix-data-prod.finance" in form_by_id[ "widget-dataset-list" ]["value"] assert ( "Description: Finance reporting marts for quarterly close and analytics." in form_by_id["widget-dataset-list"]["value"] ) assert "2. helix-data-prod.people" in form_by_id[ "widget-dataset-list" ]["value"] assert form_by_id["widget-business-reason"]["value"] == ( "Quarterly analytics review" ) submitted_form_text = "\n".join(item["value"] for item in submitted_form) assert "principal_email_group_id" not in submitted_form_text assert "compiled_preview" not in submitted_form_text assert "provider_member" not in submitted_form_text assert "{" not in submitted_form_text assert group.email in submitted_form_text request = await db_session.get(DataAccessRequest, result.request_id) assert request is not None assert request.requester_person_id == requester.id assert request.principal_email_group_id == group.id assert request.principal_google_group_id_snapshot == "google-group-analytics" assert request.principal_email_snapshot == "analytics-readers@example.com" assert request.principal_email_snapshot != "analytics-alias@example.com" assert request.feishu_instance_code == "FEISHU-DAX-001" assert request.approval_flow_key == c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT assert request.status == REQUEST_STATUS_PENDING_FEISHU assert request.trace_id == "trace-dax-016" assert request.business_reason == "Quarterly analytics review" assert request.approval_request_id == result.approval_request_id rows = ( await db_session.scalars( select(DataAccessRequestItem) .where(DataAccessRequestItem.request_id == request.id) .order_by(DataAccessRequestItem.line_no), ) ).all() assert [row.status for row in rows] == [REQUEST_ITEM_STATUS_PENDING] * 2 assert [row.resource_dataset_id_snapshot for row in rows] == [ "finance", "people", ] assert [row.resource_owner_person_id_snapshot for row in rows] == [ requester.id, requester.id, ] approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert approval is not None assert approval.external_id == "FEISHU-DAX-001" assert approval.approval_type == c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT assert approval.applicant_id == requester.id assert approval.status == "pending" source_payload = approval.source_payload assert source_payload["data_access_request_id"] == str(request.id) assert source_payload["approval_flow_key"] == c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT assert source_payload["payload_version"] == DATA_ACCESS_FEISHU_PAYLOAD_VERSION assert source_payload["semantic_intent"]["principal_email_group_id"] == ( str(group.id) ) assert source_payload["semantic_intent"]["principal_email_snapshot"] == ( "analytics-readers@example.com" ) assert source_payload["dataset_count"] == 2 assert source_payload["compiled_preview"][0]["provider_member"] == ( "group:analytics-readers@example.com" ) assert source_payload["compiled_preview"][0]["provider_role"] == ( c.BIGQUERY_DATA_VIEWER_ROLE ) assert source_payload["feishu"]["submitted_form"] == submitted_form assert "submitted_cc" not in source_payload["feishu"] assert "cc_result" not in source_payload["feishu"] serialized_source = json.dumps(source_payload, sort_keys=True) assert "app_secret" not in serialized_source assert "tenant_access_token" not in serialized_source assert "analytics-alias@example.com" not in serialized_source audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.request.submit", AuditLog.resource_id == str(request.id), ), ) assert audit is not None assert audit.actor_user_id == requester.id assert audit.status == "success" compile_audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.compile.output", AuditLog.resource_id == str(request.id), ), ) assert compile_audit is not None assert compile_audit.trace_id == audit.trace_id assert compile_audit.actor_user_id == requester.id assert compile_audit.status == "success" assert compile_audit.detail is not None assert compile_audit.detail["phase"] == "submission_preview" assert compile_audit.detail["request_id"] == str(request.id) assert compile_audit.detail["effect_count"] == 2 assert compile_audit.detail["pre_approval_preview_only"] is True assert compile_audit.detail["approval_request_id"] == str(approval.id) assert { effect["provider_member"] for effect in compile_audit.detail["effects"] } == {"group:analytics-readers@example.com"} assert { effect["desired_state"] for effect in compile_audit.detail["effects"] } == {EFFECT_DESIRED_PRESENT} assert await db_session.scalar(select(func.count()).select_from(DataAccessBinding)) == 0 assert await db_session.scalar( select(func.count()).select_from(DataAccessCompiledEffect), ) == 0 @pytest.mark.parametrize("table_id", [None, "daily_cost"]) async def test_service_account_submission_approval_and_apply_without_preregistration( db_session, employee_user, monkeypatch, table_id, ) -> None: requester, _ = employee_user resource = await _seed_resource( db_session, owner_person_id=requester.id, dataset_id="finance", ) if table_id: resource = DataAccessResource( integration_id=resource.integration_id, provider=resource.provider, resource_type=c.RESOURCE_TYPE_TABLE, project_id=resource.project_id, dataset_id=resource.dataset_id, parent_resource_id=resource.id, table_id=table_id, requestable=True, status=RESOURCE_STATUS_ACTIVE, owner_person_id=str(requester.id), owner_resolution_status=OWNER_RESOLUTION_RESOLVED, owner_source="dataset_inherited", source_version=resource.source_version, source_etag=resource.source_etag, ) db_session.add(resource) await db_session.flush() integration = await db_session.get(Integration, uuid.UUID(resource.integration_id)) integration.config_public = { "apply_mode_enabled": True, "service_account_apply_enabled": True, # No principal UUID can be known before the employee submits the email. } policy = DataAccessDatasetIamPolicy( project_id=resource.project_id, dataset_id=resource.dataset_id, table_id=table_id, etag="before", ) driver = FakeApplyDriver( policies={(resource.project_id, resource.dataset_id): policy}, table_policies={(resource.project_id, resource.dataset_id, table_id): policy}, ) async def build_driver(integration_id, *, session): assert str(integration_id) == str(integration.id) return driver monkeypatch.setattr( "app.domains.data_access.reconciler.build_driver_by_id", build_driver, ) await _seed_entitlement(db_session) await _seed_feishu_flow( db_session, flow_key=c.APPROVAL_FLOW_DATA_ACCESS_SERVICE_ACCOUNT_GRANT, payload_schema=_payload_schema(data_owner_review_node_id="data_owner_review"), ) fake_client = FakeFeishuClient() result = await DataAccessService().submit_service_account_grant_request( db_session, requester=requester, service_account_email="ETL-READER@analytics-prod.iam.gserviceaccount.com", resource_ids=[resource.id], business_reason="Run the daily finance reporting workload.", detail_url="/data-access/requests/dax-service-account", idempotency_key="dax-service-account-self-service", trace_id="trace-dax-service-account-self-service", feishu_client=fake_client, ) assert result.status == REQUEST_STATUS_PENDING_FEISHU assert result.dataset_count == 1 assert fake_client.resolved_emails == ["employee@test.com", "employee@test.com"] assert len(fake_client.created_instances) == 1 created = fake_client.created_instances[0] assert created["approval_code"] == "DATA-ACCESS-GROUP-GRANT-CODE" assert created["node_approver_user_id_list"] == [{ "key": "data_owner_review", "value": ["ou_employee"], }] submitted_form = json.loads(created["form"]) submitted_form_text = "\n".join(item["value"] for item in submitted_form) service_account_email = "etl-reader@analytics-prod.iam.gserviceaccount.com" assert service_account_email in submitted_form_text assert "technical owner" not in submitted_form_text.lower() request = await db_session.get(DataAccessRequest, result.request_id) assert request is not None assert request.principal_type_snapshot == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT assert request.principal_email_snapshot == service_account_email assert request.principal_subject_snapshot == service_account_email assert request.principal_id is not None principal = await db_session.get(DataAccessPrincipal, request.principal_id) assert principal is not None assert principal.email_snapshot == service_account_email assert principal.external_subject_id == service_account_email assert principal.catalog_id is None assert principal.source_version == "self_service_input:v1" approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert approval is not None assert approval.approval_type == c.APPROVAL_FLOW_DATA_ACCESS_SERVICE_ACCOUNT_GRANT semantic_intent = approval.source_payload["semantic_intent"] assert semantic_intent["principal_email_snapshot"] == service_account_email assert semantic_intent["principal_subject_snapshot"] == service_account_email assert semantic_intent["source_version"] == "self_service_input:v1" assert "catalog_id" not in semantic_intent assert "technical_owner_person_id" not in semantic_intent # An unapproved request must still produce no provider mutation. pending = await run_data_access_bigquery_apply_reconcile(db_session, {}) assert pending["present"]["selected_effect_count"] == 0 assert driver.set_calls == driver.table_set_calls == [] await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) request = await db_session.get(DataAccessRequest, result.request_id) assert request is not None assert request.status == REQUEST_STATUS_APPROVED bindings = ( await db_session.scalars( select(DataAccessBinding).where(DataAccessBinding.source_request_id == request.id), ) ).all() assert len(bindings) == 1 assert bindings[0].principal_id == principal.id assert bindings[0].principal_email_snapshot == service_account_email effect = await db_session.scalar( select(DataAccessCompiledEffect).where( DataAccessCompiledEffect.binding_id == bindings[0].id, ), ) assert effect is not None assert effect.provider_member == f"serviceAccount:{service_account_email}" summary = await run_data_access_bigquery_apply_reconcile(db_session, {}) assert summary["present"]["blocked_effect_count"] == 0 assert summary["present"]["applied_count"] == 1 assert effect.status == request.reconcile_status == bindings[0].reconcile_status == "applied" calls = driver.table_set_calls if table_id else driver.set_calls assert len(calls) == 1 assert (driver.set_calls if table_id else driver.table_set_calls) == [] assert calls[0][:-2] == ( (resource.project_id, resource.dataset_id, table_id) if table_id else (resource.project_id, resource.dataset_id) ) patch = calls[0][-1] assert patch.add_bindings[0].members == (f"serviceAccount:{service_account_email}",) assert patch.add_bindings[0].role == c.BIGQUERY_DATA_VIEWER_ROLE assert patch.remove_bindings == () runs = list((await db_session.scalars( select(DataAccessEffectRun).where(DataAccessEffectRun.effect_id == effect.id), )).all()) assert len(runs) == 1 assert runs[0].provider_outcome == "apply_present_applied" assert runs[0].target_table_id == table_id # Next tick cannot replay an already-applied permission. again = await run_data_access_bigquery_apply_reconcile(db_session, {}) assert again["present"]["provider_write_count"] == 0 assert len(calls) == 1 async def test_personal_submission_uses_requester_identity_and_compiles_user_member( db_session, employee_user, ) -> None: requester, _ = employee_user resource = await _seed_resource( db_session, owner_person_id=requester.id, dataset_id="personal_finance", ) await _seed_entitlement(db_session) await _seed_feishu_flow( db_session, flow_key=c.APPROVAL_FLOW_DATA_ACCESS_PERSONAL_GRANT, payload_schema=_payload_schema(data_owner_review_node_id="data_owner_review"), ) fake_client = FakeFeishuClient() result = await DataAccessService().submit_personal_grant_request( db_session, requester=requester, resource_ids=[resource.id], business_reason="Investigate the monthly finance close variance.", detail_url="/data-access/requests/dax-personal", idempotency_key="dax-personal-self-service", trace_id="trace-dax-personal-self-service", feishu_client=fake_client, ) assert result.status == REQUEST_STATUS_PENDING_FEISHU assert result.dataset_count == 1 assert fake_client.resolved_emails == ["employee@test.com", "employee@test.com"] assert len(fake_client.created_instances) == 1 created = fake_client.created_instances[0] assert created["node_approver_user_id_list"] == [{ "key": "data_owner_review", "value": ["ou_employee"], }] assert "employee@test.com" in "\n".join( item["value"] for item in json.loads(created["form"]) ) request = await db_session.get(DataAccessRequest, result.request_id) assert request is not None assert request.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER assert request.principal_email_snapshot == "employee@test.com" assert request.principal_subject_snapshot == str(requester.id) assert request.principal_id is not None principal = await db_session.get(DataAccessPrincipal, request.principal_id) assert principal is not None assert principal.principal_type == c.PRINCIPAL_TYPE_GOOGLE_USER assert principal.external_subject_id == str(requester.id) assert principal.email_snapshot == "employee@test.com" assert principal.source_version == c.GOOGLE_USER_SELF_SERVICE_SOURCE_VERSION approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert approval is not None assert approval.approval_type == c.APPROVAL_FLOW_DATA_ACCESS_PERSONAL_GRANT assert approval.source_payload["semantic_intent"] == { "domain": c.DOMAIN_DATA_ACCESS, "provider": c.PROVIDER_BIGQUERY, "resource_type": c.RESOURCE_TYPE_DATASET, "principal_type": c.PRINCIPAL_TYPE_GOOGLE_USER, "approval_flow": c.APPROVAL_FLOW_DATA_ACCESS_PERSONAL_GRANT, "principal_id": str(principal.id), "principal_subject_snapshot": str(requester.id), "principal_email_snapshot": "employee@test.com", "source_version": c.GOOGLE_USER_SELF_SERVICE_SOURCE_VERSION, "source_etag": principal.source_etag, "entitlement": c.ENTITLEMENT_READ, "entitlement_id": str(request.entitlement_id), "entitlement_version": ENTITLEMENT_VERSION_V1, } await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) binding = await db_session.scalar( select(DataAccessBinding).where(DataAccessBinding.source_request_id == request.id), ) assert binding is not None assert binding.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER assert binding.principal_id == principal.id assert binding.principal_email_snapshot == "employee@test.com" effect = await db_session.scalar( select(DataAccessCompiledEffect).where( DataAccessCompiledEffect.binding_id == binding.id, ), ) assert effect is not None assert effect.provider_member == "user:employee@test.com" assert effect.provider_access_kind == "userByEmail" # Through the guard, not by assignment. A ``before_flush`` listener refuses # a direct flip to ``left`` — and that listener is only registered once the # app has been mounted, so the bare assignment passed here in isolation and # failed the moment ANY test using the ``client`` fixture ran first. It was # green for the wrong reason; this also exercises the real transition. from app.domains.leaver_blocked.guard import LeaverBlockGuard await LeaverBlockGuard(db_session).flip_to_left( requester, via="test", trace_id=uuid.uuid4(), ) await db_session.flush() revoke = await DataAccessService().revoke_stale_personal_bindings( db_session, trace_id="trace-dax-personal-leaver", ) assert revoke.revoked_pending_count == 1 await db_session.refresh(binding) assert binding.status == "revoked_pending" absent_effect = await db_session.scalar( select(DataAccessCompiledEffect) .where( DataAccessCompiledEffect.binding_id == binding.id, DataAccessCompiledEffect.desired_state == "absent", ) .order_by(DataAccessCompiledEffect.created_at.desc()), ) assert absent_effect is not None assert absent_effect.provider_member == "user:employee@test.com" async def test_table_request_uses_exact_table_owner_and_compiles_table_effect( db_session, employee_user, ) -> None: requester, _ = employee_user table_owner = IdentityPerson( employee_no=f"P{uuid.uuid4().hex[:8]}", work_email="table-owner@test.com", display_name="Table Owner", employment_status="active", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:12]}", ) db_session.add(table_owner) dataset = await _seed_resource( db_session, owner_person_id=requester.id, dataset_id="table_granularity", ) service = DataAccessService() table = await service.resolve_requestable_bigquery_table_resource( db_session, dataset_resource_id=dataset.id, table_id="account_daily", table_type="TABLE", ) await service.reassign_bigquery_table_owner( db_session, dataset_resource_id=dataset.id, table_id="account_daily", actor=requester, new_owner_person_id=table_owner.id, ) await _seed_entitlement(db_session) await _seed_feishu_flow( db_session, flow_key=c.APPROVAL_FLOW_DATA_ACCESS_PERSONAL_GRANT, payload_schema=_payload_schema(data_owner_review_node_id="data_owner_review"), ) fake_client = FakeFeishuClient() result = await service.submit_personal_grant_request( db_session, requester=requester, resource_ids=[table.resource_id], business_reason="Read the exact daily account table.", detail_url="/data-access/requests/table-granularity", idempotency_key="dax-personal-table-granularity", trace_id="trace-dax-personal-table-granularity", feishu_client=fake_client, ) assert fake_client.resolved_emails == [ "employee@test.com", "table-owner@test.com", ] approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert approval is not None assert approval.source_payload["semantic_intent"]["resource_type"] == "table" assert approval.source_payload["dataset_owner_context"][0]["table_id"] == ( "account_daily" ) await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) effect = await db_session.scalar( select(DataAccessCompiledEffect) .join(DataAccessBinding, DataAccessBinding.id == DataAccessCompiledEffect.binding_id) .where(DataAccessBinding.source_request_id == result.request_id), ) assert effect is not None assert effect.resource_type == c.RESOURCE_TYPE_TABLE assert effect.target_project_id == "helix-data-prod" assert effect.target_dataset_id == "table_granularity" assert effect.target_table_id == "account_daily" assert effect.provider_member == "user:employee@test.com" async def test_service_account_submission_rejects_invalid_typed_email_without_record( db_session, employee_user, ) -> None: requester, _ = employee_user with pytest.raises(DataAccessSubmissionError) as exc_info: await DataAccessService().submit_service_account_grant_request( db_session, requester=requester, service_account_email="not-a-service-account@example.com", resource_ids=[], business_reason="Not reached.", detail_url="/data-access/requests/dax-service-account", idempotency_key="dax-service-account-invalid", ) assert exc_info.value.code == "DATA_ACCESS_SUBMISSION_INVALID_SERVICE_ACCOUNT_EMAIL" assert await db_session.scalar(select(func.count()).select_from(DataAccessPrincipal)) == 0 async def test_group_member_submission_ccs_group_owner_in_feishu( db_session, employee_user, admin_user, ) -> None: requester, _ = employee_user group_owner, _ = admin_user resources = [ await _seed_resource( db_session, owner_person_id=requester.id, dataset_id="finance", ), ] await _seed_entitlement(db_session) await _seed_feishu_flow(db_session) group = await _seed_group( db_session, requester_person_id=group_owner.id, add_owner=True, ) requester.__dict__["_data_access_member_group_ids_override"] = [group.id] fake_client = FakeFeishuClient() result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], entitlement=c.ENTITLEMENT_READ, business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-member-cc", idempotency_key="dax-member-cc", trace_id="trace-dax-member-cc", feishu_client=fake_client, ) assert result.status == REQUEST_STATUS_PENDING_FEISHU assert result.feishu_instance_code == "FEISHU-DAX-001" assert fake_client.resolved_emails == [ "employee@test.com", "admin@test.com", ] assert len(fake_client.created_instances) == 1 created = fake_client.created_instances[0] assert created["user_id"] == "ou_employee" assert created["node_cc_user_id_list"] is None assert fake_client.cc_instances == [ { "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", "instance_code": "FEISHU-DAX-001", "user_id": "ou_employee", "open_id": None, "cc_user_ids": ["ou_admin"], "comment": ( "FYI: a Google Group member submitted this BigQuery dataset " "access request for a group you own." ), }, ] approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert approval is not None feishu_payload = approval.source_payload["feishu"] assert feishu_payload["submitted_cc"] == { "id_type": "user_id", "cc_user_ids": ["ou_admin"], "comment": ( "FYI: a Google Group member submitted this BigQuery dataset " "access request for a group you own." ), } assert feishu_payload["cc_result"] == { "ok": True, "instance_code": "FEISHU-DAX-001", "cc_user_ids": ["ou_admin"], } submit_audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.request.submit", AuditLog.resource_id == str(result.request_id), ), ) assert submit_audit is not None assert submit_audit.detail["requester_authorized_as"] == "group_member" assert submit_audit.detail["group_owner_cc_count"] == 1 compile_audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.compile.output", AuditLog.resource_id == str(result.request_id), ), ) assert compile_audit is not None assert compile_audit.detail["requester_authorized_as"] == "group_member" assert compile_audit.detail["group_owner_cc_count"] == 1 async def test_submission_falls_back_to_open_id_when_user_id_scope_missing( db_session, employee_user, ) -> None: requester, _ = employee_user group, resources = await _valid_submission_setup(db_session, requester) fake_client = OpenIdFallbackFeishuClient() result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-open-id", idempotency_key="dax-016-open-id-fallback", trace_id="trace-dax-016-open-id-fallback", feishu_client=fake_client, ) assert result.status == REQUEST_STATUS_PENDING_FEISHU assert fake_client.resolved_emails == ["employee@test.com"] assert fake_client.resolved_open_id_emails == ["employee@test.com"] assert len(fake_client.created_instances) == 1 created = fake_client.created_instances[0] assert created["approval_code"] == "DATA-ACCESS-GROUP-GRANT-CODE" assert created["user_id"] is None assert created["open_id"] == "ou_employee" async def test_submission_sends_dynamic_resource_owner_approver_node( db_session, employee_user, admin_user, ) -> None: requester, _ = employee_user resource_owner, _ = admin_user group = await _seed_group(db_session, requester_person_id=requester.id) resources = [ await _seed_resource( db_session, owner_person_id=resource_owner.id, dataset_id="finance", ), await _seed_resource( db_session, owner_person_id=resource_owner.id, dataset_id="people", ), ] await _seed_entitlement(db_session) await _seed_feishu_flow( db_session, payload_schema=_payload_schema( data_owner_review_node_id="data_owner_review", ), ) fake_client = OpenIdFallbackFeishuClient() result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-dynamic-owner", idempotency_key="dax-016-dynamic-owner", trace_id="trace-dax-016-dynamic-owner", feishu_client=fake_client, ) assert result.status == REQUEST_STATUS_PENDING_FEISHU assert fake_client.resolved_open_id_emails == [ "employee@test.com", "admin@test.com", ] assert len(fake_client.created_instances) == 1 created = fake_client.created_instances[0] assert created["open_id"] == "ou_employee" assert created["node_approver_user_id_list"] is None assert created["node_approver_open_id_list"] == [ {"key": "data_owner_review", "value": ["ou_admin"]}, ] approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert approval is not None assert approval.source_payload["feishu"]["submitted_approvers"] == { "node_approver_open_id_list": [ {"key": "data_owner_review", "value": ["ou_admin"]}, ], } async def test_submission_with_dynamic_owner_node_fails_when_owner_not_in_feishu( db_session, employee_user, admin_user, ) -> None: requester, _ = employee_user resource_owner, _ = admin_user group = await _seed_group(db_session, requester_person_id=requester.id) resource = await _seed_resource( db_session, owner_person_id=resource_owner.id, dataset_id="finance", ) await _seed_entitlement(db_session) await _seed_feishu_flow( db_session, payload_schema=_payload_schema( data_owner_review_node_id="data_owner_review", ), ) fake_client = MissingOwnerFeishuClient() with pytest.raises(AppException) as exc_info: await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-dynamic-owner-missing", idempotency_key="dax-016-dynamic-owner-missing", trace_id="trace-dax-016-dynamic-owner-missing", feishu_client=fake_client, ) assert exc_info.value.code == "FEISHU_USER_NOT_FOUND" assert fake_client.resolved_open_id_emails == [ "employee@test.com", "admin@test.com", ] assert len(fake_client.created_instances) == 0 assert await db_session.scalar(select(func.count()).select_from(DataAccessRequest)) == 0 async def test_submission_with_dynamic_owner_node_fails_when_owner_departed( db_session, employee_user, ) -> None: """#1662: a DEPARTED dataset owner must not be routed as the authoritative approver — their Feishu account is gone/frozen so the approval would hang. Fail loud (409) so the dataset is re-owned before access is requested, mirroring the group-owner CC path which already filters to active owners.""" requester, _ = employee_user # A dedicated departed owner (constructed 'left' — the LeaverBlockGuard # only blocks FLIPPING an existing active row to 'left', not seeding one). departed_owner = IdentityPerson( id=uuid.uuid4(), work_email="departed-owner@test.com", display_name="Departed Owner", employment_status="left", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", ) db_session.add(departed_owner) await db_session.flush() group = await _seed_group(db_session, requester_person_id=requester.id) resource = await _seed_resource( db_session, owner_person_id=departed_owner.id, dataset_id="finance", ) await _seed_entitlement(db_session) await _seed_feishu_flow( db_session, payload_schema=_payload_schema( data_owner_review_node_id="data_owner_review", ), ) fake_client = OpenIdFallbackFeishuClient() with pytest.raises(AppException) as exc_info: await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-dynamic-owner-departed", idempotency_key="dax-016-dynamic-owner-departed", trace_id="trace-dax-016-dynamic-owner-departed", feishu_client=fake_client, ) assert exc_info.value.reason == "resource_owner_departed" assert exc_info.value.status_code == 409 # No durable request rows created — the submission failed loud + early. assert await db_session.scalar(select(func.count()).select_from(DataAccessRequest)) == 0 async def test_submission_idempotency_reuses_existing_request_without_feishu_replay( db_session, employee_user, ) -> None: requester, _ = employee_user _, resources = await _valid_submission_setup(db_session, requester) fake_client = FakeFeishuClient() first = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal="analytics-readers@example.com", resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-016-replay", trace_id="trace-dax-016-replay", feishu_client=fake_client, ) flow = await db_session.scalar( select(IntegrationApprovalFlow).where( IntegrationApprovalFlow.flow_key == c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, ), ) assert flow is not None flow.active = False await db_session.flush() replay = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal="analytics-readers@example.com", resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-016-replay", trace_id="trace-dax-016-replay", feishu_client=fake_client, ) assert replay.request_id == first.request_id assert replay.approval_request_id == first.approval_request_id assert replay.feishu_instance_code == first.feishu_instance_code assert len(fake_client.created_instances) == 1 assert await db_session.scalar(select(func.count()).select_from(DataAccessRequest)) == 1 assert await db_session.scalar(select(func.count()).select_from(ApprovalRequest)) == 1 async def test_submission_idempotency_true_concurrent_collision_recovers( db_engine, db_session, employee_user, ) -> None: """A true TOCTOU race (same idempotency key) must replay, not 500. Regression: the pre-check (consistent read) misses a concurrent winner that committed after this txn's REPEATABLE READ snapshot, so submission proceeds to INSERT and hits ``UNIQUE(idempotency_key)``. Two bugs made the recovery fail: (1) the row was added *before* ``begin_nested`` so the collision poisoned the outer txn (PendingRollbackError); (2) the recovery re-select was a consistent read that also couldn't see the winner. Both are fixed — this drives the real collision path. """ requester, _ = employee_user _, resources = await _valid_submission_setup(db_session, requester) resource_ids = [r.id for r in resources] await db_session.commit() svc = DataAccessService() key = "dax-016-concurrent-collision" # Pin this session's snapshot BEFORE the winner exists, so the pre-check # and recovery cannot see it via a consistent read. assert await svc._existing_request_for_idempotency_key( db_session, key, requester_person_id=requester.id, ) is None # A concurrent request (separate txn) wins and commits the row first. winner_factory = async_sessionmaker( db_engine, class_=AsyncSession, expire_on_commit=False, ) winner_client = FakeFeishuClient() async with winner_factory() as winner_db: winner = await svc.submit_group_grant_request( winner_db, requester=requester, principal="analytics-readers@example.com", resource_ids=resource_ids, business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key=key, trace_id="trace-dax-collision-winner", feishu_client=winner_client, ) await winner_db.commit() assert len(winner_client.created_instances) == 1 # Loser: pre-check misses the winner → INSERT collides → recovery. loser_client = FakeFeishuClient() loser = await svc.submit_group_grant_request( db_session, requester=requester, principal="analytics-readers@example.com", resource_ids=resource_ids, business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key=key, trace_id="trace-dax-collision-loser", feishu_client=loser_client, ) await db_session.commit() assert loser.reused_existing is True assert loser.request_id == winner.request_id assert loser.dataset_count == len(resource_ids) # Loser recovered before its own Feishu instance was created. assert len(loser_client.created_instances) == 0 # Exactly one request + one approval survived the race. assert await db_session.scalar( select(func.count()).select_from(DataAccessRequest) ) == 1 assert await db_session.scalar( select(func.count()).select_from(ApprovalRequest) ) == 1 async def test_submission_idempotency_key_is_scoped_to_requester( db_session, employee_user, admin_user, ) -> None: requester, _ = employee_user other_requester, _ = admin_user group, resources = await _valid_submission_setup(db_session, requester) fake_client = FakeFeishuClient() first = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-016-scoped-replay", trace_id="trace-dax-016-scoped-replay", feishu_client=fake_client, ) second = await DataAccessService().submit_group_grant_request( db_session, requester=other_requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-016-scoped-replay", trace_id="trace-dax-016-scoped-replay", feishu_client=fake_client, ) assert len(fake_client.created_instances) == 2 assert await db_session.scalar(select(func.count()).select_from(DataAccessRequest)) == 2 request = await db_session.get(DataAccessRequest, first.request_id) assert request is not None assert request.requester_person_id == requester.id other_request = await db_session.get(DataAccessRequest, second.request_id) assert other_request is not None assert other_request.requester_person_id == other_requester.id async def test_approved_callback_creates_bindings_effects_and_is_idempotent( db_session, employee_user, ) -> None: requester, _ = employee_user group, resources = await _valid_submission_setup(db_session, requester) result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-017-callback-approved", trace_id="trace-dax-017-callback-approved", feishu_client=FakeFeishuClient(), ) await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) request = await db_session.get(DataAccessRequest, result.request_id) approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert request is not None assert approval is not None assert request.status == REQUEST_STATUS_APPROVED assert approval.status == "approved" assert approval.source_payload["latest_instance_detail"]["status"] == "APPROVED" assert "data_access_callback_deferred_until" not in approval.source_payload assert approval.source_payload["data_access_callback_outcome"] == { "status": "approved", "request_id": str(request.id), "binding_count": 2, "effect_count": 2, "compiler_version": DATA_ACCESS_COMPILER_VERSION, } items = ( await db_session.scalars( select(DataAccessRequestItem) .where(DataAccessRequestItem.request_id == request.id) .order_by(DataAccessRequestItem.line_no), ) ).all() assert [item.status for item in items] == [REQUEST_ITEM_STATUS_APPROVED] * 2 bindings = ( await db_session.scalars( select(DataAccessBinding) .where(DataAccessBinding.source_request_id == request.id) .order_by(DataAccessBinding.principal_email_snapshot), ) ).all() assert len(bindings) == 2 assert {binding.status for binding in bindings} == {BINDING_STATUS_ACTIVE} assert {binding.resource_id for binding in bindings} == { resource.id for resource in resources } assert {binding.principal_email_group_id for binding in bindings} == {group.id} assert {binding.principal_google_group_id_snapshot for binding in bindings} == { "google-group-analytics", } assert {binding.principal_email_snapshot for binding in bindings} == { "analytics-readers@example.com", } effects = ( await db_session.scalars( select(DataAccessCompiledEffect) .join( DataAccessBinding, DataAccessBinding.id == DataAccessCompiledEffect.binding_id, ) .where(DataAccessBinding.source_request_id == request.id) .order_by(DataAccessCompiledEffect.target_dataset_id), ) ).all() assert len(effects) == 2 assert [effect.target_dataset_id for effect in effects] == [ "finance", "people", ] assert {effect.provider_member for effect in effects} == { "group:analytics-readers@example.com", } assert {effect.provider_role for effect in effects} == { c.BIGQUERY_DATA_VIEWER_ROLE, } assert {effect.desired_state for effect in effects} == {EFFECT_DESIRED_PRESENT} assert {effect.managed_reason for effect in effects} == { EFFECT_MANAGED_REASON_GRANT_REQUEST, } assert {effect.status for effect in effects} == {EFFECT_STATUS_PENDING_RECONCILE} callback_trace_id = AuditService._coerce_trace_id( f"data_access_callback:{result.feishu_instance_code}", ) callback_received = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.request.callback_received", AuditLog.resource_id == str(request.id), ), ) assert callback_received is not None assert callback_received.trace_id == callback_trace_id assert callback_received.actor_type == "integration" assert callback_received.actor_user_id == requester.id assert callback_received.detail is not None assert callback_received.detail["approval_status"] == "approved" assert callback_received.detail["feishu_instance_code"] == ( result.feishu_instance_code ) binding_audits = ( await db_session.scalars( select(AuditLog) .where(AuditLog.action == "data_access.binding.created") .order_by(AuditLog.resource_id), ) ).all() assert len(binding_audits) == 2 assert {audit.trace_id for audit in binding_audits} == {callback_trace_id} assert { audit.detail["request_id"] for audit in binding_audits if audit.detail is not None } == {str(request.id)} assert { audit.detail["binding_status"] for audit in binding_audits if audit.detail is not None } == {BINDING_STATUS_ACTIVE} assert {audit.resource_id for audit in binding_audits} == { str(binding.id) for binding in bindings } callback_compile_audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.compile.output", AuditLog.resource_id == str(request.id), AuditLog.trace_id == callback_trace_id, ), ) assert callback_compile_audit is not None assert callback_compile_audit.detail is not None assert callback_compile_audit.detail["phase"] == "approval_callback" assert callback_compile_audit.detail["effect_count"] == 2 assert callback_compile_audit.detail["binding_count"] == 2 assert { effect["effect_id"] for effect in callback_compile_audit.detail["effects"] } == {str(effect.id) for effect in effects} assert { effect["binding_id"] for effect in callback_compile_audit.detail["effects"] } == {str(binding.id) for binding in bindings} assert { effect["desired_state"] for effect in callback_compile_audit.detail["effects"] } == {EFFECT_DESIRED_PRESENT} await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) assert await db_session.scalar(select(func.count()).select_from(DataAccessBinding)) == 2 assert await db_session.scalar( select(func.count()).select_from(DataAccessCompiledEffect), ) == 2 assert await db_session.scalar( select(func.count()) .select_from(AuditLog) .where( AuditLog.action == "data_access.request.approved", AuditLog.resource_id == str(request.id), ), ) == 1 assert await db_session.scalar( select(func.count()) .select_from(AuditLog) .where( AuditLog.action == "data_access.request.callback_received", AuditLog.resource_id == str(request.id), ), ) == 1 assert await db_session.scalar( select(func.count()) .select_from(AuditLog) .where(AuditLog.action == "data_access.binding.created"), ) == 2 assert await db_session.scalar( select(func.count()) .select_from(AuditLog) .where( AuditLog.action == "data_access.compile.output", AuditLog.trace_id == callback_trace_id, ), ) == 1 async def test_rejected_callback_settles_request_without_bindings_or_effects( db_session, employee_user, ) -> None: requester, _ = employee_user group, resources = await _valid_submission_setup(db_session, requester) result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-017-callback-rejected", trace_id="trace-dax-017-callback-rejected", feishu_client=FakeFeishuClient(), ) await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "REJECTED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) request = await db_session.get(DataAccessRequest, result.request_id) approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert request is not None assert approval is not None assert request.status == REQUEST_STATUS_REJECTED assert approval.status == "rejected" assert approval.source_payload["data_access_callback_outcome"]["status"] == ( REQUEST_STATUS_REJECTED ) item_statuses = ( await db_session.scalars( select(DataAccessRequestItem.status) .where(DataAccessRequestItem.request_id == request.id) .order_by(DataAccessRequestItem.line_no), ) ).all() assert item_statuses == [REQUEST_ITEM_STATUS_REJECTED] * 2 assert await db_session.scalar(select(func.count()).select_from(DataAccessBinding)) == 0 assert await db_session.scalar( select(func.count()).select_from(DataAccessCompiledEffect), ) == 0 rejected_trace_id = AuditService._coerce_trace_id( f"data_access_callback:{result.feishu_instance_code}", ) callback_received = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.request.callback_received", AuditLog.resource_id == str(request.id), ), ) assert callback_received is not None assert callback_received.trace_id == rejected_trace_id assert callback_received.detail is not None assert callback_received.detail["approval_status"] == "rejected" rejected_audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.request.rejected", AuditLog.resource_id == str(request.id), ), ) assert rejected_audit is not None assert rejected_audit.trace_id == rejected_trace_id async def test_audit_receipt_is_idempotent_and_compile_output_chunks( db_session, employee_user, ) -> None: requester, _ = employee_user group, resources = await _valid_submission_setup(db_session, requester) result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-020-audit-helper", trace_id="trace-dax-020-audit-helper", feishu_client=FakeFeishuClient(), ) request = await db_session.get(DataAccessRequest, result.request_id) approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert request is not None assert approval is not None service = DataAccessService() callback_trace = f"data_access_callback:{result.feishu_instance_code}" for _ in range(2): await service._audit_callback_received( db_session, approval_request=approval, request=request, instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", }, source_payload=approval.source_payload, approval_status="approved", trace_id=callback_trace, ) assert await db_session.scalar( select(func.count()) .select_from(AuditLog) .where( AuditLog.action == "data_access.request.callback_received", AuditLog.resource_id == str(request.id), ), ) == 1 effect_count = 18 effects = [ SimpleNamespace( id=uuid.uuid4(), binding_id=uuid.uuid4(), resource_id=uuid.uuid4(), principal_email_group_id=uuid.uuid4(), entitlement_id=uuid.uuid4(), provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_DATASET, target_project_id="p" * 64, target_dataset_id=f"{index:02d}_" + ("d" * 1021), provider_member=c.BIGQUERY_GROUP_MEMBER_PREFIX + ("a" * 314), provider_role=c.BIGQUERY_DATA_VIEWER_ROLE, desired_state=EFFECT_DESIRED_PRESENT, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, resource_source_version_snapshot="v" * 128, resource_source_etag_snapshot=f"{index:02d}_" + ("e" * 253), compiler_version=DATA_ACCESS_COMPILER_VERSION, managed_reason=EFFECT_MANAGED_REASON_GRANT_REQUEST, already_present_at_first_apply=False, status=EFFECT_STATUS_PENDING_RECONCILE, ) for index in range(effect_count) ] await service._audit_compile_output( db_session, trace_id="trace-dax-020-large-compile", actor_type="system", actor_user_id=None, resource_type="data_access_request", resource_id="large-compile-output", phase="approval_callback", effects=effects, request_id=request.id, detail_extra={"approval_request_id": str(approval.id)}, ) compile_rows = ( await db_session.scalars( select(AuditLog) .where( AuditLog.action == "data_access.compile.output", AuditLog.resource_id == "large-compile-output", ) ) ).all() compile_rows = sorted( compile_rows, key=lambda row: row.detail["effect_chunk_index"] if row.detail else 0, ) assert len(compile_rows) > 1 assert [ row.detail["effect_chunk_index"] for row in compile_rows if row.detail is not None ] == list(range(1, len(compile_rows) + 1)) assert { row.detail["effect_count"] for row in compile_rows if row.detail is not None } == {effect_count} assert { row.detail["effect_chunk_count"] for row in compile_rows if row.detail is not None } == {len(compile_rows)} assert all( row.detail is not None and "_truncated" not in row.detail and len(json.dumps(row.detail, default=str)) <= DATA_ACCESS_AUDIT_DETAIL_SOFT_LIMIT_BYTES for row in compile_rows ) assert { effect["effect_id"] for row in compile_rows if row.detail is not None for effect in row.detail["effects"] } == {str(effect.id) for effect in effects} async def test_approved_callback_allows_same_google_group_primary_email_rename( db_session, employee_user, ) -> None: requester, _ = employee_user group, resources = await _valid_submission_setup(db_session, requester) result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-017-callback-rename", trace_id="trace-dax-017-callback-rename", feishu_client=FakeFeishuClient(), ) group.email = "analytics-renamed@example.com" await db_session.flush() await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) request = await db_session.get(DataAccessRequest, result.request_id) assert request is not None assert request.status == REQUEST_STATUS_APPROVED bindings = ( await db_session.scalars( select(DataAccessBinding).where( DataAccessBinding.source_request_id == request.id, ), ) ).all() assert {binding.principal_email_snapshot for binding in bindings} == { "analytics-renamed@example.com", } effects = ( await db_session.scalars( select(DataAccessCompiledEffect) .join( DataAccessBinding, DataAccessBinding.id == DataAccessCompiledEffect.binding_id, ) .where(DataAccessBinding.source_request_id == request.id), ) ).all() assert {effect.provider_member for effect in effects} == { "group:analytics-renamed@example.com", } async def test_approved_callback_bridges_deleted_group_to_active_same_google_id( db_session, employee_user, ) -> None: requester, _ = employee_user group, resources = await _valid_submission_setup(db_session, requester) result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-018-callback-deleted-bridge", trace_id="trace-dax-018-callback-deleted-bridge", feishu_client=FakeFeishuClient(), ) group.status = "deleted" active_bridge = EmailGroup( email="analytics-bridge-new@example.com", name="Analytics Bridge New", status="active", google_group_id=group.google_group_id, ) db_session.add(active_bridge) await db_session.flush() db_session.add( EmailGroupOwner( email_group_id=active_bridge.id, person_id=requester.id, ), ) await db_session.flush() await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) request = await db_session.get(DataAccessRequest, result.request_id) assert request is not None assert request.status == REQUEST_STATUS_APPROVED bindings = ( await db_session.scalars( select(DataAccessBinding).where( DataAccessBinding.source_request_id == request.id, ), ) ).all() assert {binding.principal_email_group_id for binding in bindings} == { active_bridge.id, } assert {binding.principal_email_snapshot for binding in bindings} == { "analytics-bridge-new@example.com", } effects = ( await db_session.scalars( select(DataAccessCompiledEffect) .join( DataAccessBinding, DataAccessBinding.id == DataAccessCompiledEffect.binding_id, ) .where(DataAccessBinding.source_request_id == request.id), ) ).all() assert {effect.provider_member for effect in effects} == { "group:analytics-bridge-new@example.com", } async def test_callback_refuses_stale_resource_owner_without_bindings_or_effects( db_session, employee_user, admin_user, ) -> None: requester, _ = employee_user new_owner, _ = admin_user group, resources = await _valid_submission_setup(db_session, requester) result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-017-callback-stale-owner", trace_id="trace-dax-017-callback-stale-owner", feishu_client=FakeFeishuClient(), ) resources[0].owner_person_id = str(new_owner.id) await db_session.flush() await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) request = await db_session.get(DataAccessRequest, result.request_id) approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert request is not None assert approval is not None assert request.status == REQUEST_STATUS_STALE_REQUIRES_RESUBMIT assert approval.status == "approved" assert approval.source_payload["data_access_callback_refusal"]["reason"] == ( "resource_context_changed" ) rows = ( await db_session.scalars( select(DataAccessRequestItem).where( DataAccessRequestItem.request_id == request.id, ), ) ).all() assert {row.status for row in rows} == {REQUEST_ITEM_STATUS_STALE} assert {row.staleness_reason for row in rows} == {"resource_context_changed"} assert await db_session.scalar(select(func.count()).select_from(DataAccessBinding)) == 0 assert await db_session.scalar( select(func.count()).select_from(DataAccessCompiledEffect), ) == 0 audit = await db_session.scalar( select(AuditLog).where( AuditLog.action == "data_access.request.callback_refused", AuditLog.resource_id == str(request.id), ), ) assert audit is not None assert audit.status == "failure" assert audit.trace_id == AuditService._coerce_trace_id( f"data_access_callback:{result.feishu_instance_code}", ) assert audit.detail is not None assert audit.detail["reason"] == "resource_context_changed" @pytest.mark.parametrize( ("case", "expected_reason", "expected_field"), [ ("requestability", "resource_resource_not_requestable", None), ( "source_version", "resource_catalog_version_stale", "source_version", ), ( "source_etag", "resource_catalog_version_stale", "source_etag", ), ], ) async def test_callback_refuses_requestability_or_catalog_version_drift( db_session, employee_user, case: str, expected_reason: str, expected_field: str | None, ) -> None: requester, _ = employee_user group, resources = await _valid_submission_setup(db_session, requester) result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key=f"dax-018-callback-{case}", trace_id=f"trace-dax-018-callback-{case}", feishu_client=FakeFeishuClient(), ) if case == "requestability": resources[0].requestable = False elif case == "source_version": resources[0].source_version = "catalog-2026-06-11-finance" else: resources[0].source_etag = "etag-finance-v2" await db_session.flush() await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) request = await db_session.get(DataAccessRequest, result.request_id) approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert request is not None assert approval is not None assert request.status == REQUEST_STATUS_STALE_REQUIRES_RESUBMIT refusal = approval.source_payload["data_access_callback_refusal"] assert refusal["reason"] == expected_reason if expected_field is not None: assert refusal["field"] == expected_field rows = ( await db_session.scalars( select(DataAccessRequestItem).where( DataAccessRequestItem.request_id == request.id, ), ) ).all() assert {row.status for row in rows} == {REQUEST_ITEM_STATUS_STALE} assert {row.staleness_reason for row in rows} == {expected_reason} assert await db_session.scalar(select(func.count()).select_from(DataAccessBinding)) == 0 assert await db_session.scalar( select(func.count()).select_from(DataAccessCompiledEffect), ) == 0 async def test_callback_refuses_duplicate_google_group_identity( db_session, employee_user, ) -> None: requester, _ = employee_user group, resources = await _valid_submission_setup(db_session, requester) result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-017-callback-duplicate-group", trace_id="trace-dax-017-callback-duplicate-group", feishu_client=FakeFeishuClient(), ) db_session.add( EmailGroup( email="analytics-duplicate@example.com", name="Analytics Duplicate", status="active", google_group_id=group.google_group_id, ), ) await db_session.flush() await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) request = await db_session.get(DataAccessRequest, result.request_id) approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert request is not None assert approval is not None assert request.status == REQUEST_STATUS_STALE_REQUIRES_RESUBMIT assert approval.source_payload["data_access_callback_refusal"]["reason"] == ( "principal_duplicate_active_google_group_id" ) assert await db_session.scalar(select(func.count()).select_from(DataAccessBinding)) == 0 assert await db_session.scalar( select(func.count()).select_from(DataAccessCompiledEffect), ) == 0 async def test_callback_refuses_missing_google_group_id( db_session, employee_user, ) -> None: requester, _ = employee_user group, resources = await _valid_submission_setup(db_session, requester) result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-018-callback-missing-google-id", trace_id="trace-dax-018-callback-missing-google-id", feishu_client=FakeFeishuClient(), ) group.google_group_id = None await db_session.flush() await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) request = await db_session.get(DataAccessRequest, result.request_id) approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert request is not None assert approval is not None assert request.status == REQUEST_STATUS_STALE_REQUIRES_RESUBMIT assert approval.source_payload["data_access_callback_refusal"]["reason"] == ( "principal_missing_google_group_id" ) assert await db_session.scalar(select(func.count()).select_from(DataAccessBinding)) == 0 assert await db_session.scalar( select(func.count()).select_from(DataAccessCompiledEffect), ) == 0 @pytest.mark.parametrize( ("case", "expected_reason"), [ ("deleted_group", "principal_deleted_group"), ("stale_entitlement", "entitlement_stale"), ("stale_entitlement_version", "entitlement_stale"), ("stale_requester", "requester_stale"), ], ) async def test_callback_refuses_stale_group_entitlement_or_requester_state( db_session, employee_user, case: str, expected_reason: str, ) -> None: requester, _ = employee_user group, resources = await _valid_submission_setup(db_session, requester) result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key=f"dax-017-callback-{case}", trace_id=f"trace-dax-017-callback-{case}", feishu_client=FakeFeishuClient(), ) if case == "deleted_group": group.status = "deleted" elif case == "stale_entitlement": entitlement = await db_session.scalar(select(DataAccessEntitlement)) assert entitlement is not None entitlement.requestable = False elif case == "stale_entitlement_version": request = await db_session.get(DataAccessRequest, result.request_id) assert request is not None request.entitlement_version_snapshot = "v0" else: requester.employment_status = "leaver_pending" await db_session.flush() await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) request = await db_session.get(DataAccessRequest, result.request_id) approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert request is not None assert approval is not None assert request.status == REQUEST_STATUS_STALE_REQUIRES_RESUBMIT assert approval.source_payload["data_access_callback_refusal"]["reason"] == ( expected_reason ) assert await db_session.scalar(select(func.count()).select_from(DataAccessBinding)) == 0 assert await db_session.scalar( select(func.count()).select_from(DataAccessCompiledEffect), ) == 0 async def test_callback_refuses_stale_compiler_version_without_bindings_or_effects( db_session, employee_user, ) -> None: requester, _ = employee_user group, resources = await _valid_submission_setup(db_session, requester) result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-018-callback-stale-compiler", trace_id="trace-dax-018-callback-stale-compiler", feishu_client=FakeFeishuClient(), ) approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert approval is not None source_payload = dict(approval.source_payload) compiled_preview = [ dict(preview) for preview in source_payload["compiled_preview"] ] compiled_preview[0]["compiler_version"] = "dax-compiler-v0" source_payload["compiled_preview"] = compiled_preview approval.source_payload = source_payload await db_session.flush() await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) request = await db_session.get(DataAccessRequest, result.request_id) assert request is not None assert request.status == REQUEST_STATUS_STALE_REQUIRES_RESUBMIT assert approval.source_payload["data_access_callback_refusal"]["reason"] == ( "compiler_version_stale" ) assert approval.source_payload["data_access_callback_refusal"][ "callback_compiler_version" ] == "dax-compiler-v0" assert await db_session.scalar(select(func.count()).select_from(DataAccessBinding)) == 0 assert await db_session.scalar( select(func.count()).select_from(DataAccessCompiledEffect), ) == 0 async def test_callback_refuses_feishu_instance_request_mismatch( db_session, employee_user, ) -> None: requester, _ = employee_user group, resources = await _valid_submission_setup(db_session, requester) result = await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key="dax-017-callback-mismatch", trace_id="trace-dax-017-callback-mismatch", feishu_client=FakeFeishuClient(), ) request = await db_session.get(DataAccessRequest, result.request_id) assert request is not None request.feishu_instance_code = "FEISHU-DAX-DIFFERENT" await db_session.flush() await ApprovalService(db_session).handle_feishu_callback( {"event": {"instance_code": result.feishu_instance_code}}, prefetched_instance_detail={ "instance_code": result.feishu_instance_code, "status": "APPROVED", "approval_code": "DATA-ACCESS-GROUP-GRANT-CODE", }, ) approval = await db_session.get(ApprovalRequest, result.approval_request_id) assert approval is not None assert request.status == REQUEST_STATUS_STALE_REQUIRES_RESUBMIT assert approval.source_payload["data_access_callback_refusal"]["reason"] == ( "approval_external_id_mismatch" ) assert await db_session.scalar(select(func.count()).select_from(DataAccessBinding)) == 0 assert await db_session.scalar( select(func.count()).select_from(DataAccessCompiledEffect), ) == 0 @pytest.mark.parametrize( ("case", "expected_code"), [ ("missing_flow", "APPROVAL_FLOW_NOT_CONFIGURED"), ("inactive_flow", "APPROVAL_FLOW_NOT_CONFIGURED"), ("invalid_resource", "DATA_ACCESS_NON_FEDERATED_RESOURCE"), ("deleted_group", "DATA_ACCESS_DELETED_GROUP"), ("unauthorized_requester", "DATA_ACCESS_REQUESTER_NOT_AUTHORIZED"), ("raw_provider_role", "DATA_ACCESS_SUBMISSION_UNSUPPORTED_ENTITLEMENT_INPUT"), ("table_resource", "DATA_ACCESS_UNSUPPORTED_RESOURCE_TYPE"), ("project_resource", "DATA_ACCESS_UNSUPPORTED_RESOURCE_TYPE"), ("user_principal", "DATA_ACCESS_UNSUPPORTED_PRINCIPAL_TYPE"), ("person_principal", "DATA_ACCESS_UNSUPPORTED_PRINCIPAL_TYPE"), ], ) async def test_submission_rejects_security_bar_cases_before_feishu( db_session, employee_user, case: str, expected_code: str, ) -> None: requester, _ = employee_user if case == "missing_flow": await _seed_feishu_flow(db_session, flow_key="cloud_scope_request") else: await _seed_feishu_flow(db_session, active=case != "inactive_flow") await _seed_entitlement(db_session) add_owner = case != "unauthorized_requester" group = await _seed_group( db_session, requester_person_id=requester.id, status="deleted" if case == "deleted_group" else "active", add_owner=add_owner, ) resource_id: uuid.UUID if case == "invalid_resource": resource_id = uuid.uuid4() else: resource_type = c.RESOURCE_TYPE_DATASET if case == "table_resource": resource_type = "table" elif case == "project_resource": resource_type = "project" resource = await _seed_resource( db_session, owner_person_id=requester.id, dataset_id=f"dataset_{case}", resource_type=resource_type, ) resource_id = resource.id fake_client = FakeFeishuClient() principal_type = c.PRINCIPAL_TYPE_GOOGLE_GROUP if case == "user_principal": principal_type = "user" elif case == "person_principal": principal_type = "person" with pytest.raises(AppException) as exc_info: await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal=group.id, principal_type=principal_type, resource_ids=[resource_id], entitlement=( c.BIGQUERY_DATA_VIEWER_ROLE if case == "raw_provider_role" else c.ENTITLEMENT_READ ), business_reason="Quarterly analytics review", detail_url="/data-access/requests/dax-draft", idempotency_key=f"dax-016-reject-{case}", trace_id=f"trace-dax-016-reject-{case}", feishu_client=fake_client, ) assert exc_info.value.code == expected_code assert fake_client.resolved_emails == [] assert fake_client.created_instances == [] assert await db_session.scalar(select(func.count()).select_from(DataAccessRequest)) == 0 assert await db_session.scalar(select(func.count()).select_from(ApprovalRequest)) == 0 async def _submit_fault_case( db_session: AsyncSession, *, requester: IdentityPerson, resources: list[DataAccessResource], client: FakeFeishuClient, key: str, ): return await DataAccessService().submit_group_grant_request( db_session, requester=requester, principal="analytics-readers@example.com", resource_ids=[resource.id for resource in resources], business_reason="Quarterly analytics durable submission", detail_url="/data-access/requests/durable", idempotency_key=key, trace_id=f"trace-{key}", feishu_client=client, ) async def test_response_loss_60012_replay_reuses_request_and_uuid( db_session, employee_user, ) -> None: requester, _ = employee_user _, resources = await _valid_submission_setup(db_session, requester) fake_client = ResponseLossFeishuClient() with pytest.raises(AppException) as caught: await _submit_fault_case( db_session, requester=requester, resources=resources, client=fake_client, key="dax-response-loss", ) assert caught.value.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN" replay = await _submit_fault_case( db_session, requester=requester, resources=resources, client=fake_client, key="dax-response-loss", ) assert replay.reused_existing is True assert replay.feishu_instance_code == fake_client.instance_code assert len(fake_client.created_instances) == 2 assert { call["instance_uuid"] for call in fake_client.created_instances } == {fake_client.created_instances[0]["instance_uuid"]} assert await db_session.scalar( select(func.count()).select_from(DataAccessRequest) ) == 1 assert await db_session.scalar( select(func.count()).select_from(ApprovalRequest) ) == 1 async def test_process_exit_after_provider_acceptance_replays_without_repost( db_session, employee_user, monkeypatch, ) -> None: requester, _ = employee_user _, resources = await _valid_submission_setup(db_session, requester) fake_client = FakeFeishuClient() original = DataAccessService._finalize_group_grant_submission async def _exit(*args, **kwargs): raise SystemExit("worker terminated after provider acceptance") monkeypatch.setattr(DataAccessService, "_finalize_group_grant_submission", _exit) with pytest.raises(SystemExit): await _submit_fault_case( db_session, requester=requester, resources=resources, client=fake_client, key="dax-process-exit", ) intent = (await db_session.scalars(select(ApprovalSubmissionIntent))).one() assert intent.state == "provider_created" monkeypatch.setattr( DataAccessService, "_finalize_group_grant_submission", original, ) replay = await _submit_fault_case( db_session, requester=requester, resources=resources, client=fake_client, key="dax-process-exit", ) assert replay.feishu_instance_code == "FEISHU-DAX-001" assert len(fake_client.created_instances) == 1 await db_session.refresh(intent) assert intent.state == "completed" async def test_scheduled_reconciler_recovers_unknown_create( db_session, employee_user, ) -> None: requester, _ = employee_user _, resources = await _valid_submission_setup(db_session, requester) fake_client = ResponseLossFeishuClient() with pytest.raises(AppException): await _submit_fault_case( db_session, requester=requester, resources=resources, client=fake_client, key="dax-cron-recovery", ) await make_submission_intent_retries_due(db_session) summary = await DataAccessService().reconcile_submission_intents( db_session, min_age_minutes=0, feishu_client=fake_client, ) assert summary == { "checked": 1, "reconciled": 1, "still_unknown": 0, "failed": 0, } request = (await db_session.scalars(select(DataAccessRequest))).one() assert request.feishu_instance_code == fake_client.instance_code async def test_definite_provider_rejection_marks_request_and_items_terminal( db_session, employee_user, ) -> None: requester, _ = employee_user _, resources = await _valid_submission_setup(db_session, requester) fake_client = RejectingFeishuClient() with pytest.raises(AppException) as caught: await _submit_fault_case( db_session, requester=requester, resources=resources, client=fake_client, key="dax-definite-rejection", ) assert caught.value.code == "FEISHU_REQUEST_REJECTED" request = (await db_session.scalars(select(DataAccessRequest))).one() items = list( (await db_session.scalars(select(DataAccessRequestItem))).all() ) intent = (await db_session.scalars(select(ApprovalSubmissionIntent))).one() assert request.status == REQUEST_STATUS_REJECTED assert {item.status for item in items} == {REQUEST_ITEM_STATUS_REJECTED} assert intent.state == "failed" assert await db_session.scalar( select(func.count()).select_from(ApprovalRequest) ) == 0 async def test_definite_rejection_during_unknown_replay_closes_request_items( db_session, employee_user, ) -> None: requester, _ = employee_user _, resources = await _valid_submission_setup(db_session, requester) fake_client = UnknownThenRejectingFeishuClient() with pytest.raises(AppException): await _submit_fault_case( db_session, requester=requester, resources=resources, client=fake_client, key="dax-replay-definite-rejection", ) with pytest.raises(AppException) as caught: await _submit_fault_case( db_session, requester=requester, resources=resources, client=fake_client, key="dax-replay-definite-rejection", ) assert caught.value.code == "FEISHU_REQUEST_REJECTED" request = (await db_session.scalars(select(DataAccessRequest))).one() items = list((await db_session.scalars(select(DataAccessRequestItem))).all()) assert request.status == REQUEST_STATUS_REJECTED assert {item.status for item in items} == {REQUEST_ITEM_STATUS_REJECTED}