from __future__ import annotations

import shutil
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace

from fastapi import APIRouter, Depends, FastAPI
from fastapi.testclient import TestClient

from govoplan_access.auth import ApiPrincipal, get_api_principal
from govoplan_core.core.access import (
    ACCESS_CAPABILITY_NAMES,
    ACCESS_MODULE_ID,
    CAPABILITY_ACCESS_ADMINISTRATION,
    CAPABILITY_ACCESS_DIRECTORY,
    CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER,
    CAPABILITY_ACCESS_PERMISSION_EVALUATOR,
    CAPABILITY_ACCESS_PRINCIPAL_RESOLVER,
    CAPABILITY_ACCESS_RESOURCE_ACCESS,
    CAPABILITY_ACCESS_EXPLANATION,
    CAPABILITY_ACCESS_SEMANTIC_DIRECTORY,
    CAPABILITY_ACCESS_TENANT_PROVISIONER,
    CAPABILITY_AUDIT_RECORDER,
    CAPABILITY_AUDIT_RETENTION,
    CAPABILITY_AUDIT_SINK,
    CAPABILITY_AUTH_API_PRINCIPAL_PROVIDER,
    CAPABILITY_TENANCY_TENANT_RESOLVER,
    AccessDecision,
    AccessAdministration,
    AccessDirectory,
    AccessDecisionProvenance,
    AccessExplanationService,
    AccessGovernanceMaterializer,
    ResourceAccessExplanationProvider,
    AccessSemanticDirectory,
    AccessSubjectRef,
    AccountRef,
    AuditEvent,
    AuditRecordRef,
    AuditRecorder,
    AuditRetentionProvider,
    AuditSink,
    CreatedApiKeyRef,
    DevelopmentBootstrapRef,
    FunctionAssignmentRef,
    FunctionDelegationRef,
    FunctionRef,
    GovernanceTemplateMaterialization,
    GroupRef,
    IdentityRef,
    OrganizationUnitRef,
    PermissionEvaluator,
    PrincipalRef,
    PrincipalResolver,
    ResourceAccessService,
    RoleRef,
    TenantRef,
    TenantAccessProvisioner,
    TenantOwnerCandidateRef,
    TenantResolver,
    UserRef,
)
from govoplan_core.security.secrets import CAPABILITY_SECURITY_SECRET_PROVIDER, SecretProvider
from govoplan_core.core.campaigns import (
    CAPABILITY_CAMPAIGNS_ACCESS,
    CAPABILITY_CAMPAIGNS_DELIVERY_TASKS,
    CAPABILITY_CAMPAIGNS_MAIL_POLICY_CONTEXT,
    CAPABILITY_CAMPAIGNS_POLICY_CONTEXT,
    CAPABILITY_CAMPAIGNS_RETENTION,
    CampaignAccessProvider,
    CampaignDeliveryTaskProvider,
    CampaignMailPolicyContext,
    CampaignMailPolicyContextProvider,
    CampaignPolicyContext,
    CampaignPolicyContextProvider,
    CampaignRetentionProvider,
)
from govoplan_core.core.files import CAPABILITY_FILES_ACCESS, FileAccessProvider
from govoplan_core.core.modules import ModuleContext, ModuleManifest
from govoplan_core.core.idm import CAPABILITY_IDM_DIRECTORY, OrganizationFunctionAssignmentRef
from govoplan_core.core.organizations import CAPABILITY_ORGANIZATION_DIRECTORY, OrganizationDirectory as CoreOrganizationDirectory, OrganizationFunctionRef as CoreOrganizationFunctionRef, OrganizationUnitRef as CoreOrganizationUnitRef
from govoplan_core.core.registry import PlatformRegistry
from govoplan_core.core.runtime import clear_runtime, configure_runtime
from govoplan_core.db.base import Base
from govoplan_core.server.config import GovoplanServerConfig
from govoplan_core.tenancy.scope import create_scope_tables
from govoplan_access.backend.db.models import (
    Account,
    ExternalFunctionRoleAssignment,
    Function,
    FunctionAssignment,
    FunctionRoleAssignment,
    Group,
    Identity,
    IdentityAccountLink,
    OrganizationUnit,
    Role,
    User,
    UserGroupMembership,
)
from govoplan_tenancy.backend.db.models import Tenant
from tests.db_isolation import temporary_database


class _FakeAccessDirectory:
    account = AccountRef(id="account-1", email="demo@example.test", display_name="Demo")
    user = UserRef(id="user-1", account_id=account.id, tenant_id="tenant-1", email=account.email)
    group = GroupRef(id="group-1", tenant_id="tenant-1", name="Team")

    def get_account(self, account_id: str) -> AccountRef | None:
        return self.account if account_id == self.account.id else None

    def get_user(self, user_id: str) -> UserRef | None:
        return self.user if user_id == self.user.id else None

    def get_users(self, user_ids):
        return {self.user.id: self.user} if self.user.id in set(user_ids) else {}

    def users_for_tenant(self, tenant_id: str):
        return (self.user,) if tenant_id == self.user.tenant_id else ()

    def get_group(self, group_id: str) -> GroupRef | None:
        return self.group if group_id == self.group.id else None

    def get_groups(self, group_ids):
        return {self.group.id: self.group} if self.group.id in set(group_ids) else {}

    def groups_for_tenant(self, tenant_id: str):
        return (self.group,) if tenant_id == self.group.tenant_id else ()

    def groups_for_user(self, user_id: str, *, tenant_id: str):
        if user_id == self.user.id and tenant_id == self.user.tenant_id:
            return (self.group,)
        return ()

    def display_label(self, subject: AccessSubjectRef) -> str | None:
        if subject.kind == "user" and subject.id == self.user.id:
            return self.user.display_name or self.user.email
        if subject.kind == "group" and subject.id == self.group.id:
            return self.group.name
        return subject.label


class _FakeAccessSemanticDirectory(_FakeAccessDirectory):
    identity = IdentityRef(
        id="identity-1",
        display_name="Demo Person",
        primary_account_id=_FakeAccessDirectory.account.id,
        account_ids=(_FakeAccessDirectory.account.id, "account-2"),
    )
    organization_unit = OrganizationUnitRef(id="ou-1", tenant_id="tenant-1", name="Registry Office")
    function = FunctionRef(
        id="function-1",
        tenant_id="tenant-1",
        organization_unit_id=organization_unit.id,
        slug="registry-clerk",
        name="Registry Clerk",
        role_ids=("role-1",),
        delegable=True,
        act_in_place_allowed=True,
    )
    assignment = FunctionAssignmentRef(
        id="assignment-1",
        tenant_id="tenant-1",
        account_id=_FakeAccessDirectory.account.id,
        identity_id=identity.id,
        function_id=function.id,
        organization_unit_id=organization_unit.id,
        applies_to_subunits=True,
    )

    def get_identity(self, identity_id: str) -> IdentityRef | None:
        return self.identity if identity_id == self.identity.id else None

    def accounts_for_identity(self, identity_id: str):
        return (self.account,) if identity_id == self.identity.id else ()

    def get_organization_unit(self, organization_unit_id: str) -> OrganizationUnitRef | None:
        return self.organization_unit if organization_unit_id == self.organization_unit.id else None

    def organization_units_for_tenant(self, tenant_id: str):
        return (self.organization_unit,) if tenant_id == self.organization_unit.tenant_id else ()

    def get_function(self, function_id: str) -> FunctionRef | None:
        return self.function if function_id == self.function.id else None

    def functions_for_organization_unit(self, organization_unit_id: str, *, include_subunits: bool = False):
        del include_subunits
        return (self.function,) if organization_unit_id == self.organization_unit.id else ()

    def get_function_assignment(self, assignment_id: str) -> FunctionAssignmentRef | None:
        return self.assignment if assignment_id == self.assignment.id else None

    def function_assignments_for_account(self, account_id: str, *, tenant_id: str | None = None):
        if account_id != self.account.id:
            return ()
        if tenant_id is not None and tenant_id != self.assignment.tenant_id:
            return ()
        return (self.assignment,)


class _FakePrincipalResolver:
    def resolve_request(self, request: object, *, session: object | None = None) -> PrincipalRef:
        del request, session
        return PrincipalRef(
            account_id="account-1",
            membership_id="user-1",
            tenant_id="tenant-1",
            scopes=frozenset({"files:file:read"}),
            group_ids=frozenset({"group-1"}),
        )


class _FakeTenantResolver:
    tenant = TenantRef(id="tenant-1", name="Tenant")

    def get_tenant(self, tenant_id: str) -> TenantRef | None:
        return self.tenant if tenant_id == self.tenant.id else None

    def current_tenant(self, principal: PrincipalRef) -> TenantRef | None:
        return self.get_tenant(principal.tenant_id or "")


class _FakePermissionEvaluator:
    def scopes_grant(self, scopes, required_scope: str) -> bool:
        return required_scope in set(scopes)

    def has_scope(self, principal: PrincipalRef, required_scope: str) -> bool:
        return self.scopes_grant(principal.scopes, required_scope)

    def explain_scope(self, principal: PrincipalRef, required_scope: str) -> AccessDecision:
        allowed = self.has_scope(principal, required_scope)
        return AccessDecision(allowed=allowed, reason=None if allowed else "missing scope", requirements=(required_scope,))


class _FakeResourceAccessService:
    def explain(self, principal: PrincipalRef, *, resource_type: str, resource_id: str, action: str) -> AccessDecision:
        del principal, resource_type, resource_id, action
        return AccessDecision(allowed=True)


class _FakeResourceAccessExplanationProvider:
    def explain_resource_provenance(
        self,
        session: object,
        principal: PrincipalRef,
        *,
        resource_type: str,
        resource_id: str,
        action: str,
    ):
        del session, principal, resource_type, action
        return (AccessDecisionProvenance(kind="resource", id=resource_id, label="Demo resource"),)


class _FakeFileAccessProvider(_FakeResourceAccessExplanationProvider):
    pass


class _FakeAccessExplanationService:
    provenance = (
        AccessDecisionProvenance(kind="identity", id="identity-1", label="Demo Person"),
        AccessDecisionProvenance(kind="function", id="assignment-1", label="Registry Clerk", tenant_id="tenant-1"),
        AccessDecisionProvenance(kind="role", id="role-1", label="Clerk", tenant_id="tenant-1"),
        AccessDecisionProvenance(kind="right", id="files:file:read", label="Read files"),
    )

    def explain_scope_provenance(self, principal: PrincipalRef, required_scope: str):
        del principal, required_scope
        return self.provenance

    def explain_resource_provenance(
        self,
        principal: PrincipalRef,
        *,
        resource_type: str,
        resource_id: str,
        action: str,
    ):
        del principal, resource_type, resource_id, action
        return self.provenance


class _FakeTenantAccessProvisioner:
    def ensure_default_roles(self, session: object, tenant: object | None = None):
        del session, tenant
        return {}

    def tenant_owner_candidates(self, session: object):
        del session
        return (TenantOwnerCandidateRef(account_id="account-1", email="demo@example.test", display_name="Demo"),)

    def ensure_tenant_owner_membership(self, session: object, *, tenant: object, owner_account_id: str) -> str:
        del session, tenant, owner_account_id
        return "user-1"

    def ensure_development_admin(self, session: object, *, tenant: object, user_email: str, user_password: str, api_key_secret: str | None, scopes):
        del session, tenant, user_password, scopes
        return DevelopmentBootstrapRef(
            user=UserRef(id="user-1", account_id="account-1", tenant_id="tenant-1", email=user_email),
            created_api_key=CreatedApiKeyRef(id="api-key-1", secret=api_key_secret) if api_key_secret else None,
        )


class _FakeAccessAdministration:
    def tenant_counts(self, session: object, tenant_id: str):
        del session, tenant_id
        return {"users": 1, "active_users": 1, "groups": 2, "api_keys": 3}

    def system_account_count(self, session: object) -> int:
        del session
        return 1

    def role_count_for_tenant(self, session: object, tenant_id: str) -> int:
        del session, tenant_id
        return 2

    def active_api_key_count_for_tenant(self, session: object, tenant_id: str) -> int:
        del session, tenant_id
        return 3

    def actor_email_by_user_id(self, session: object, user_ids):
        del session
        return {user_id: "demo@example.test" for user_id in user_ids}

    def user_ids_for_actor_filter(self, session: object, *, operator: str, value: str):
        del session, operator, value
        return ("user-1",)

    def user_settings(self, session: object, user_id: str, *, tenant_id: str):
        del session, user_id, tenant_id
        return {"privacy_retention_policy": {"audit_detail_level": "minimal"}}

    def set_user_settings(self, session: object, user_id: str, *, tenant_id: str, settings):
        del session, user_id, tenant_id
        return dict(settings)

    def group_settings(self, session: object, group_id: str, *, tenant_id: str):
        del session, group_id, tenant_id
        return {"privacy_retention_policy": {"audit_detail_level": "redacted"}}

    def set_group_settings(self, session: object, group_id: str, *, tenant_id: str, settings):
        del session, group_id, tenant_id
        return dict(settings)


class _FakeAccessGovernanceMaterializer:
    def __init__(self) -> None:
        self.synced: list[GovernanceTemplateMaterialization] = []
        self.removed: list[GovernanceTemplateMaterialization] = []

    def sync_template(self, session: object, template: GovernanceTemplateMaterialization) -> None:
        del session
        self.synced.append(template)

    def remove_template(self, session: object, template: GovernanceTemplateMaterialization) -> None:
        del session
        self.removed.append(template)


class _FakeCampaignMailPolicyContextProvider:
    context = CampaignMailPolicyContext(
        id="campaign-1",
        tenant_id="tenant-1",
        owner_user_id="user-1",
        owner_group_id="group-1",
        mail_profile_policy={"allow_campaign_profiles": False},
    )

    def get_campaign_mail_policy_context(self, session: object, *, tenant_id: str, campaign_id: str):
        del session
        if tenant_id == self.context.tenant_id and campaign_id == self.context.id:
            return self.context
        return None

    def set_campaign_mail_profile_policy(self, session: object, *, tenant_id: str, campaign_id: str, policy):
        del session
        if tenant_id != self.context.tenant_id or campaign_id != self.context.id:
            raise ValueError("Campaign policy not found")
        return dict(policy)


class _FakeCampaignAccessProvider:
    def campaign_exists(self, session: object, *, tenant_id: str, campaign_id: str) -> bool:
        del session
        return tenant_id == "tenant-1" and campaign_id == "campaign-1"

    def can_read_campaign(
        self,
        session: object,
        *,
        tenant_id: str,
        campaign_id: str,
        user_id: str,
        group_ids=(),
        tenant_admin: bool = False,
    ) -> bool:
        del session
        return self.campaign_exists(object(), tenant_id=tenant_id, campaign_id=campaign_id) and (
            tenant_admin or user_id == "user-1" or "group-1" in set(group_ids)
        )


class _FakeCampaignPolicyContextProvider:
    context = CampaignPolicyContext(
        id="campaign-1",
        tenant_id="tenant-1",
        owner_user_id="user-1",
        owner_group_id="group-1",
        settings={"privacy_retention_policy": {"audit_detail_level": "redacted"}},
    )

    def get_campaign_policy_context(self, session: object, *, campaign_id: str, tenant_id: str | None = None):
        del session
        if campaign_id != self.context.id or (tenant_id is not None and tenant_id != self.context.tenant_id):
            return None
        return self.context

    def set_campaign_settings(self, session: object, *, tenant_id: str, campaign_id: str, settings):
        del session
        if tenant_id != self.context.tenant_id or campaign_id != self.context.id:
            raise ValueError("Campaign privacy policy not found")
        return dict(settings)


class _FakeCampaignDeliveryTaskProvider:
    def send_campaign_job(self, session: object, *, job_id: str, enqueue_imap_task: bool = True):
        del session
        return {"job_id": job_id, "enqueue_imap_task": enqueue_imap_task}

    def append_sent_for_job(self, session: object, *, job_id: str):
        del session
        return {"job_id": job_id, "status": "appended"}


class _FakeCampaignRetentionProvider:
    def apply_retention(self, session: object, *, dry_run: bool, now, policy_for_campaign_id):
        del session, now
        policy_for_campaign_id("campaign-1")
        return {"raw_campaign_json": {"eligible": int(dry_run)}}


class _FakeSecretProvider:
    def __init__(self) -> None:
        self._values: dict[str, str] = {}

    def store_secret(self, *, scope: str, name: str, value: str) -> str:
        ref = f"{scope}/{name}"
        self._values[ref] = value
        return ref

    def read_secret(self, secret_ref: str) -> str | None:
        return self._values.get(secret_ref)

    def delete_secret(self, secret_ref: str) -> None:
        self._values.pop(secret_ref, None)


class _FakeAuditSink:
    def __init__(self) -> None:
        self.events: list[AuditEvent] = []

    def record(self, event: AuditEvent) -> None:
        self.events.append(event)


class _FakeAuditRecorder:
    def record_event(self, session: object, event: AuditEvent) -> AuditRecordRef:
        del session
        return AuditRecordRef(
            id="audit-1",
            scope=event.scope,
            tenant_id=event.tenant_id,
            user_id=event.user_id,
            api_key_id=event.api_key_id,
            action=event.action,
            object_type=event.resource_type,
            object_id=event.resource_id,
            details=event.details,
        )


class _FakeAuditRetentionProvider:
    def apply_detail_retention(self, session: object, *, dry_run: bool, now, policy_for_record):
        del session, dry_run, now, policy_for_record
        return {"eligible": 1, "redacted": 0, "already_redacted": 0}


class AccessContractTests(unittest.TestCase):
    def test_access_capability_names_are_stable(self) -> None:
        self.assertEqual("access", ACCESS_MODULE_ID)
        self.assertEqual("access.principalResolver", CAPABILITY_ACCESS_PRINCIPAL_RESOLVER)
        self.assertEqual("access.directory", CAPABILITY_ACCESS_DIRECTORY)
        self.assertEqual("access.permissionEvaluator", CAPABILITY_ACCESS_PERMISSION_EVALUATOR)
        self.assertEqual("access.resourceAccess", CAPABILITY_ACCESS_RESOURCE_ACCESS)
        self.assertEqual("access.semanticDirectory", CAPABILITY_ACCESS_SEMANTIC_DIRECTORY)
        self.assertEqual("access.explanation", CAPABILITY_ACCESS_EXPLANATION)
        self.assertEqual("access.tenantProvisioner", CAPABILITY_ACCESS_TENANT_PROVISIONER)
        self.assertEqual("access.administration", CAPABILITY_ACCESS_ADMINISTRATION)
        self.assertEqual("access.governanceMaterializer", CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER)
        self.assertEqual("campaigns.access", CAPABILITY_CAMPAIGNS_ACCESS)
        self.assertEqual("files.access", CAPABILITY_FILES_ACCESS)
        self.assertEqual("campaigns.deliveryTasks", CAPABILITY_CAMPAIGNS_DELIVERY_TASKS)
        self.assertEqual("campaigns.mailPolicyContext", CAPABILITY_CAMPAIGNS_MAIL_POLICY_CONTEXT)
        self.assertEqual("campaigns.policyContext", CAPABILITY_CAMPAIGNS_POLICY_CONTEXT)
        self.assertEqual("campaigns.retention", CAPABILITY_CAMPAIGNS_RETENTION)
        self.assertEqual("tenancy.tenantResolver", CAPABILITY_TENANCY_TENANT_RESOLVER)
        self.assertEqual("security.secretProvider", CAPABILITY_SECURITY_SECRET_PROVIDER)
        self.assertEqual("audit.sink", CAPABILITY_AUDIT_SINK)
        self.assertEqual("audit.recorder", CAPABILITY_AUDIT_RECORDER)
        self.assertEqual("audit.retention", CAPABILITY_AUDIT_RETENTION)
        self.assertEqual(len(ACCESS_CAPABILITY_NAMES), len(set(ACCESS_CAPABILITY_NAMES)))

    def test_access_refs_are_small_immutable_dtos(self) -> None:
        principal = PrincipalRef(
            account_id="account-1",
            membership_id="user-1",
            tenant_id="tenant-1",
            identity_id="identity-1",
            scopes=frozenset({"campaigns:campaign:read"}),
            group_ids=frozenset({"group-1"}),
            role_ids=frozenset({"role-1"}),
            function_assignment_ids=frozenset({"assignment-1"}),
            delegation_ids=frozenset({"delegation-1"}),
            acting_for_account_id="account-2",
            email="demo@example.test",
        )
        role = RoleRef(id="role-1", name="Reviewer", level="tenant", tenant_id="tenant-1")

        self.assertEqual("account-1", principal.account_id)
        self.assertEqual(frozenset({"campaigns:campaign:read"}), principal.scopes)
        self.assertEqual("Reviewer", role.name)
        payload = principal.to_dict()
        self.assertEqual(["campaigns:campaign:read"], payload["scopes"])
        self.assertEqual(["group-1"], payload["group_ids"])
        self.assertEqual(["role-1"], payload["role_ids"])
        self.assertEqual(["assignment-1"], payload["function_assignment_ids"])
        self.assertEqual(["delegation-1"], payload["delegation_ids"])
        self.assertEqual("identity-1", payload["identity_id"])
        self.assertEqual("account-2", payload["acting_for_account_id"])
        self.assertEqual("session", payload["auth_method"])
        self.assertIsNone(payload["api_key_id"])
        self.assertIsNone(payload["service_account_id"])
        self.assertEqual(principal, PrincipalRef.from_mapping(payload))
        with self.assertRaises(AttributeError):
            principal.account_id = "changed"  # type: ignore[misc]

    def test_identity_function_and_delegation_refs_model_organization_scope(self) -> None:
        identity = IdentityRef(
            id="identity-1",
            display_name="Demo Person",
            primary_account_id="account-1",
            account_ids=("account-1", "account-2"),
        )
        organization_unit = OrganizationUnitRef(id="ou-1", tenant_id="tenant-1", name="Registry Office")
        function = FunctionRef(
            id="function-1",
            tenant_id="tenant-1",
            organization_unit_id=organization_unit.id,
            slug="registry-clerk",
            name="Registry Clerk",
            role_ids=("role-1",),
            delegable=True,
            act_in_place_allowed=True,
        )
        assignment = FunctionAssignmentRef(
            id="assignment-1",
            tenant_id="tenant-1",
            account_id="account-1",
            identity_id=identity.id,
            function_id=function.id,
            organization_unit_id=organization_unit.id,
            applies_to_subunits=True,
        )
        delegation = FunctionDelegationRef(
            id="delegation-1",
            tenant_id="tenant-1",
            function_assignment_id=assignment.id,
            delegator_account_id="account-1",
            delegate_account_id="account-2",
            mode="act_in_place",
        )
        provenance = AccessDecisionProvenance(
            kind="function",
            id=assignment.id,
            label=function.name,
            tenant_id="tenant-1",
            source="delegation",
            details={"organization_unit_id": organization_unit.id, "applies_to_subunits": True},
        )

        self.assertEqual(("account-1", "account-2"), identity.account_ids)
        self.assertEqual("ou-1", function.organization_unit_id)
        self.assertTrue(assignment.applies_to_subunits)
        self.assertEqual("act_in_place", delegation.mode)
        self.assertEqual("function", provenance.to_dict()["kind"])
        self.assertEqual({"organization_unit_id": "ou-1", "applies_to_subunits": True}, provenance.to_dict()["details"])

    def test_protocol_shapes_match_reference_implementations(self) -> None:
        self.assertIsInstance(_FakePrincipalResolver(), PrincipalResolver)
        self.assertIsInstance(_FakeTenantResolver(), TenantResolver)
        self.assertIsInstance(_FakeAccessDirectory(), AccessDirectory)
        self.assertIsInstance(_FakeAccessSemanticDirectory(), AccessSemanticDirectory)
        self.assertIsInstance(_FakePermissionEvaluator(), PermissionEvaluator)
        self.assertIsInstance(_FakeResourceAccessService(), ResourceAccessService)
        self.assertIsInstance(_FakeResourceAccessExplanationProvider(), ResourceAccessExplanationProvider)
        self.assertIsInstance(_FakeFileAccessProvider(), FileAccessProvider)
        self.assertIsInstance(_FakeAccessExplanationService(), AccessExplanationService)
        self.assertIsInstance(_FakeTenantAccessProvisioner(), TenantAccessProvisioner)
        self.assertIsInstance(_FakeAccessAdministration(), AccessAdministration)
        self.assertIsInstance(_FakeAccessGovernanceMaterializer(), AccessGovernanceMaterializer)
        self.assertIsInstance(_FakeCampaignAccessProvider(), CampaignAccessProvider)
        self.assertIsInstance(_FakeCampaignDeliveryTaskProvider(), CampaignDeliveryTaskProvider)
        self.assertIsInstance(_FakeCampaignMailPolicyContextProvider(), CampaignMailPolicyContextProvider)
        self.assertIsInstance(_FakeCampaignPolicyContextProvider(), CampaignPolicyContextProvider)
        self.assertIsInstance(_FakeCampaignRetentionProvider(), CampaignRetentionProvider)
        self.assertIsInstance(_FakeSecretProvider(), SecretProvider)
        self.assertIsInstance(_FakeAuditSink(), AuditSink)
        self.assertIsInstance(_FakeAuditRecorder(), AuditRecorder)
        self.assertIsInstance(_FakeAuditRetentionProvider(), AuditRetentionProvider)

    def test_campaign_mail_policy_context_capability_shape(self) -> None:
        provider = _FakeCampaignMailPolicyContextProvider()
        context = provider.get_campaign_mail_policy_context(object(), tenant_id="tenant-1", campaign_id="campaign-1")

        self.assertEqual("campaign-1", context.id)  # type: ignore[union-attr]
        self.assertEqual("user-1", context.owner_user_id)  # type: ignore[union-attr]
        self.assertEqual({"allow_campaign_profiles": False}, context.mail_profile_policy)  # type: ignore[union-attr]
        self.assertEqual({"allow_campaign_profiles": True}, provider.set_campaign_mail_profile_policy(object(), tenant_id="tenant-1", campaign_id="campaign-1", policy={"allow_campaign_profiles": True}))

    def test_campaign_access_capability_shape(self) -> None:
        provider = _FakeCampaignAccessProvider()

        self.assertTrue(provider.campaign_exists(object(), tenant_id="tenant-1", campaign_id="campaign-1"))
        self.assertTrue(provider.can_read_campaign(object(), tenant_id="tenant-1", campaign_id="campaign-1", user_id="other", group_ids=("group-1",)))
        self.assertFalse(provider.can_read_campaign(object(), tenant_id="tenant-1", campaign_id="missing", user_id="user-1"))

    def test_campaign_policy_delivery_and_retention_capability_shapes(self) -> None:
        policy_provider = _FakeCampaignPolicyContextProvider()
        delivery_provider = _FakeCampaignDeliveryTaskProvider()
        retention_provider = _FakeCampaignRetentionProvider()

        context = policy_provider.get_campaign_policy_context(object(), tenant_id="tenant-1", campaign_id="campaign-1")
        self.assertEqual("tenant-1", context.tenant_id)  # type: ignore[union-attr]
        self.assertEqual({"privacy_retention_policy": {"audit_detail_level": "redacted"}}, context.settings)  # type: ignore[union-attr]
        self.assertEqual({"demo": True}, policy_provider.set_campaign_settings(object(), tenant_id="tenant-1", campaign_id="campaign-1", settings={"demo": True}))
        self.assertEqual({"job_id": "job-1", "enqueue_imap_task": False}, delivery_provider.send_campaign_job(object(), job_id="job-1", enqueue_imap_task=False))
        self.assertEqual({"job_id": "job-1", "status": "appended"}, delivery_provider.append_sent_for_job(object(), job_id="job-1"))
        self.assertEqual({"raw_campaign_json": {"eligible": 1}}, retention_provider.apply_retention(object(), dry_run=True, now=object(), policy_for_campaign_id=lambda campaign_id: object()))

    def test_access_capabilities_register_and_resolve_through_platform_registry(self) -> None:
        directory = _FakeAccessDirectory()
        semantic_directory = _FakeAccessSemanticDirectory()
        explanation = _FakeAccessExplanationService()
        resolver = _FakePrincipalResolver()
        evaluator = _FakePermissionEvaluator()
        tenant_resolver = _FakeTenantResolver()
        administration = _FakeAccessAdministration()
        materializer = _FakeAccessGovernanceMaterializer()
        manifest = ModuleManifest(
            id=ACCESS_MODULE_ID,
            name="Access",
            version="test",
            capability_factories={
                CAPABILITY_ACCESS_DIRECTORY: lambda context: directory,
                CAPABILITY_ACCESS_SEMANTIC_DIRECTORY: lambda context: semantic_directory,
                CAPABILITY_ACCESS_EXPLANATION: lambda context: explanation,
                CAPABILITY_ACCESS_PRINCIPAL_RESOLVER: lambda context: resolver,
                CAPABILITY_ACCESS_PERMISSION_EVALUATOR: lambda context: evaluator,
                CAPABILITY_ACCESS_ADMINISTRATION: lambda context: administration,
                CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER: lambda context: materializer,
                CAPABILITY_TENANCY_TENANT_RESOLVER: lambda context: tenant_resolver,
            },
        )

        registry = PlatformRegistry()
        registry.register(manifest)
        registry.configure_capability_context(ModuleContext(registry=registry, settings=object()))

        self.assertIs(directory, registry.require_capability(CAPABILITY_ACCESS_DIRECTORY))
        self.assertIs(semantic_directory, registry.require_capability(CAPABILITY_ACCESS_SEMANTIC_DIRECTORY))
        self.assertIs(explanation, registry.require_capability(CAPABILITY_ACCESS_EXPLANATION))
        self.assertIs(resolver, registry.require_capability(CAPABILITY_ACCESS_PRINCIPAL_RESOLVER))
        self.assertIs(evaluator, registry.require_capability(CAPABILITY_ACCESS_PERMISSION_EVALUATOR))
        self.assertIs(administration, registry.require_capability(CAPABILITY_ACCESS_ADMINISTRATION))
        self.assertIs(materializer, registry.require_capability(CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER))
        self.assertIs(tenant_resolver, registry.require_capability(CAPABILITY_TENANCY_TENANT_RESOLVER))

    def test_sql_directory_and_tenant_resolver_return_access_dtos(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-access-directory-"))
        try:
            with temporary_database(f"sqlite:///{root / 'directory.db'}") as database:
                create_scope_tables(database.engine)
                Base.metadata.create_all(bind=database.engine)
                with database.session() as session:
                    tenant = Tenant(id="tenant-directory", slug="directory", name="Directory Tenant", settings={})
                    account = Account(
                        id="account-directory",
                        email="directory@example.test",
                        normalized_email="directory@example.test",
                        display_name="Directory User",
                    )
                    user = User(
                        id="user-directory",
                        tenant_id=tenant.id,
                        account_id=account.id,
                        email=account.email,
                        display_name=None,
                        settings={},
                        mail_profile_policy={},
                    )
                    group = Group(id="group-directory", tenant_id=tenant.id, slug="review", name="Review", description=None)
                    membership = UserGroupMembership(tenant_id=tenant.id, user_id=user.id, group_id=group.id)
                    identity = Identity(id="identity-directory", display_name="Directory Person", source="local")
                    identity_link = IdentityAccountLink(
                        identity_id=identity.id,
                        account_id=account.id,
                        is_primary=True,
                        source="local",
                    )
                    organization_unit = OrganizationUnit(
                        id="ou-directory",
                        tenant_id=tenant.id,
                        slug="registry",
                        name="Registry",
                    )
                    role = Role(
                        id="role-directory",
                        tenant_id=tenant.id,
                        slug="registry-reader",
                        name="Registry Reader",
                        permissions=["files:read"],
                    )
                    function = Function(
                        id="function-directory",
                        tenant_id=tenant.id,
                        organization_unit_id=organization_unit.id,
                        slug="registry-clerk",
                        name="Registry Clerk",
                        delegable=True,
                    )
                    function_role = FunctionRoleAssignment(tenant_id=tenant.id, function_id=function.id, role_id=role.id)
                    function_assignment = FunctionAssignment(
                        id="assignment-directory",
                        tenant_id=tenant.id,
                        account_id=account.id,
                        identity_id=identity.id,
                        function_id=function.id,
                        organization_unit_id=organization_unit.id,
                        applies_to_subunits=True,
                    )
                    session.add_all([
                        tenant,
                        account,
                        user,
                        group,
                        membership,
                        identity,
                        identity_link,
                        organization_unit,
                        role,
                        function,
                        function_role,
                        function_assignment,
                    ])
                    session.commit()

                from govoplan_access.backend.directory import SqlAccessDirectory
                from govoplan_access.backend.security.sessions import collect_user_scopes
                from govoplan_tenancy.backend.capabilities import SqlTenantResolver

                directory = SqlAccessDirectory()
                tenant_resolver = SqlTenantResolver()

                self.assertEqual("directory@example.test", directory.get_account("account-directory").email)  # type: ignore[union-attr]
                self.assertEqual("Directory User", directory.get_user("user-directory").display_name)  # type: ignore[union-attr]
                self.assertEqual(["user-directory"], [user.id for user in directory.users_for_tenant("tenant-directory")])
                self.assertEqual(["group-directory"], [group.id for group in directory.groups_for_tenant("tenant-directory")])
                self.assertEqual(["group-directory"], [group.id for group in directory.groups_for_user("user-directory", tenant_id="tenant-directory")])
                self.assertEqual("Review", directory.display_label(AccessSubjectRef(kind="group", id="group-directory", tenant_id="tenant-directory")))
                self.assertEqual("Directory Person", directory.get_identity("identity-directory").display_name)  # type: ignore[union-attr]
                self.assertEqual(["account-directory"], [account.id for account in directory.accounts_for_identity("identity-directory")])
                self.assertEqual("Registry", directory.get_organization_unit("ou-directory").name)  # type: ignore[union-attr]
                self.assertEqual(["role-directory"], list(directory.get_function("function-directory").role_ids))  # type: ignore[union-attr]
                self.assertEqual(
                    ["assignment-directory"],
                    [item.id for item in directory.function_assignments_for_account("account-directory", tenant_id="tenant-directory")],
                )
                with database.session() as session:
                    user = session.get(User, "user-directory")
                    self.assertIn("files:read", collect_user_scopes(session, user, include_system=False))  # type: ignore[arg-type]
                self.assertEqual("directory", tenant_resolver.get_tenant("tenant-directory").slug)  # type: ignore[union-attr]
                self.assertEqual(
                    "tenant-directory",
                    tenant_resolver.current_tenant(
                        PrincipalRef(account_id="account-directory", membership_id="user-directory", tenant_id="tenant-directory")
                    ).id,  # type: ignore[union-attr]
                )
        finally:
            shutil.rmtree(root, ignore_errors=True)

    def test_semantic_access_admin_crud_endpoints(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-access-semantic-api-"))
        try:
            with temporary_database(f"sqlite:///{root / 'semantic.db'}") as database:
                create_scope_tables(database.engine)
                Base.metadata.create_all(bind=database.engine)
                tenant_id = "tenant-semantic"
                account_id = "account-semantic"
                delegate_account_id = "account-semantic-delegate"
                user_id = "user-semantic"
                role_id = "role-semantic"
                with database.session() as session:
                    tenant = Tenant(id=tenant_id, slug="semantic", name="Semantic Tenant", settings={})
                    account = Account(
                        id=account_id,
                        email="semantic@example.test",
                        normalized_email="semantic@example.test",
                        display_name="Semantic User",
                    )
                    delegate_account = Account(
                        id=delegate_account_id,
                        email="delegate@example.test",
                        normalized_email="delegate@example.test",
                        display_name="Delegate User",
                    )
                    user = User(
                        id=user_id,
                        tenant_id=tenant_id,
                        account_id=account_id,
                        email=account.email,
                        display_name=account.display_name,
                        settings={},
                        mail_profile_policy={},
                    )
                    role = Role(
                        id=role_id,
                        tenant_id=tenant_id,
                        slug="semantic-reader",
                        name="Semantic Reader",
                        permissions=["files:read"],
                    )
                    session.add_all([tenant, account, delegate_account, user, role])
                    session.commit()

                from govoplan_access.backend.api.v1.routes import router as admin_router

                app = FastAPI()
                app.include_router(admin_router)

                def principal_override() -> ApiPrincipal:
                    with database.session() as session:
                        account = session.get(Account, account_id)
                        user = session.get(User, user_id)
                        assert account is not None
                        assert user is not None
                        return ApiPrincipal(
                            principal=PrincipalRef(
                                account_id=account_id,
                                membership_id=user_id,
                                tenant_id=tenant_id,
                                scopes=frozenset({
                                    "admin:roles:read",
                                    "admin:roles:write",
                                    "admin:roles:assign",
                                    "system:accounts:read",
                                    "system:accounts:update",
                                }),
                            ),
                            account=account,
                            user=user,
                        )

                app.dependency_overrides[get_api_principal] = principal_override

                with TestClient(app) as client:
                    identity_response = client.post(
                        "/admin/identities",
                        json={"display_name": "Semantic Person", "account_ids": [account_id], "primary_account_id": account_id},
                    )
                    self.assertEqual(201, identity_response.status_code, identity_response.text)
                    identity_id = identity_response.json()["id"]

                    unit_response = client.post(
                        "/admin/organization-units",
                        json={"slug": "registry", "name": "Registry"},
                    )
                    self.assertEqual(201, unit_response.status_code, unit_response.text)
                    organization_unit_id = unit_response.json()["id"]

                    function_response = client.post(
                        "/admin/functions",
                        json={
                            "organization_unit_id": organization_unit_id,
                            "slug": "registry-clerk",
                            "name": "Registry Clerk",
                            "role_ids": [role_id],
                            "delegable": True,
                        },
                    )
                    self.assertEqual(201, function_response.status_code, function_response.text)
                    function_id = function_response.json()["id"]
                    self.assertEqual([role_id], function_response.json()["role_ids"])

                    assignment_response = client.post(
                        "/admin/function-assignments",
                        json={
                            "account_id": account_id,
                            "identity_id": identity_id,
                            "function_id": function_id,
                            "applies_to_subunits": True,
                        },
                    )
                    self.assertEqual(201, assignment_response.status_code, assignment_response.text)
                    assignment_id = assignment_response.json()["id"]
                    self.assertTrue(assignment_response.json()["applies_to_subunits"])

                    delegation_response = client.post(
                        "/admin/function-delegations",
                        json={"function_assignment_id": assignment_id, "delegate_account_id": delegate_account_id},
                    )
                    self.assertEqual(201, delegation_response.status_code, delegation_response.text)
                    self.assertEqual(delegate_account_id, delegation_response.json()["delegate_account_id"])

                    list_response = client.get(f"/admin/function-assignments?account_id={account_id}")
                    self.assertEqual(200, list_response.status_code, list_response.text)
                    self.assertEqual([assignment_id], [item["id"] for item in list_response.json()["assignments"]])
        finally:
            shutil.rmtree(root, ignore_errors=True)

    def test_external_function_role_mappings_require_organizations_directory(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-access-external-function-map-"))
        try:
            tenant_id = "tenant-external-function-map"
            account_id = "account-external-function-map"
            user_id = "user-external-function-map"
            role_id = "role-external-function-map"
            function_id = "function-external-function-map"

            with temporary_database(f"sqlite:///{root / 'external-function-map.db'}") as database:
                create_scope_tables(database.engine)
                Base.metadata.create_all(bind=database.engine)
                with database.session() as session:
                    tenant = Tenant(id=tenant_id, slug="external-function-map", name="External Function Map", settings={})
                    account = Account(
                        id=account_id,
                        email="external-function-map@example.test",
                        normalized_email="external-function-map@example.test",
                        display_name="External Function Map Admin",
                    )
                    user = User(
                        id=user_id,
                        tenant_id=tenant_id,
                        account_id=account_id,
                        email=account.email,
                        display_name=account.display_name,
                        settings={},
                        mail_profile_policy={},
                    )
                    role = Role(
                        id=role_id,
                        tenant_id=tenant_id,
                        slug="external-function-reader",
                        name="External Function Reader",
                        permissions=["files:file:read"],
                        is_assignable=True,
                    )
                    session.add_all([tenant, account, user, role])
                    session.commit()

                from govoplan_access.backend.api.v1.routes import router as admin_router

                app = FastAPI()
                app.include_router(admin_router)

                def principal_override() -> ApiPrincipal:
                    with database.session() as session:
                        account = session.get(Account, account_id)
                        user = session.get(User, user_id)
                        assert account is not None
                        assert user is not None
                        return ApiPrincipal(
                            principal=PrincipalRef(
                                account_id=account_id,
                                membership_id=user_id,
                                tenant_id=tenant_id,
                                scopes=frozenset({"admin:roles:read", "admin:roles:write", "access:role:assign"}),
                            ),
                            account=account,
                            user=user,
                        )

                app.dependency_overrides[get_api_principal] = principal_override
                clear_runtime()

                with TestClient(app) as client:
                    blocked_response = client.post(
                        "/admin/external-function-role-mappings",
                        json={"source_module": "organizations", "function_id": function_id, "role_id": role_id},
                    )
                    self.assertEqual(409, blocked_response.status_code, blocked_response.text)
                    self.assertEqual("organizations_required", blocked_response.json()["detail"]["code"])

                    class FakeOrganizationDirectory(CoreOrganizationDirectory):
                        def get_organization_unit(self, organization_unit_id: str):
                            del organization_unit_id
                            return None

                        def organization_units_for_tenant(self, tenant_id: str):
                            del tenant_id
                            return ()

                        def get_function(self, requested_function_id: str):
                            if requested_function_id != function_id:
                                return None
                            return CoreOrganizationFunctionRef(
                                id=function_id,
                                tenant_id=tenant_id,
                                organization_unit_id="unit-external-function-map",
                                slug="external-function",
                                name="External Function",
                            )

                        def functions_for_organization_unit(self, organization_unit_id: str, *, include_subunits: bool = False):
                            del organization_unit_id, include_subunits
                            return ()

                    registry = PlatformRegistry()
                    registry.register(
                        ModuleManifest(
                            id="organizations",
                            name="Organizations",
                            version="test",
                            capability_factories={CAPABILITY_ORGANIZATION_DIRECTORY: lambda context: FakeOrganizationDirectory()},
                        )
                    )
                    context = ModuleContext(registry=registry, settings=object())
                    registry.configure_capability_context(context)
                    configure_runtime(context)

                    created_response = client.post(
                        "/admin/external-function-role-mappings",
                        json={"source_module": "organizations", "function_id": function_id, "role_id": role_id},
                    )
                    self.assertEqual(201, created_response.status_code, created_response.text)
                    self.assertEqual(function_id, created_response.json()["function_id"])

                    unsupported_response = client.post(
                        "/admin/external-function-role-mappings",
                        json={"source_module": "legacy", "function_id": function_id, "role_id": role_id},
                    )
                    self.assertEqual(422, unsupported_response.status_code, unsupported_response.text)
        finally:
            clear_runtime()
            shutil.rmtree(root, ignore_errors=True)

    def test_access_explanation_includes_idm_function_role_sources(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-access-explanation-"))
        try:
            tenant_id = "tenant-access-explanation"
            account_id = "account-access-explanation"
            user_id = "user-access-explanation"
            role_id = "role-access-explanation"
            function_id = "function-access-explanation"
            organization_unit_id = "unit-access-explanation"
            assignment_id = "assignment-access-explanation"
            identity_id = "identity-access-explanation"

            assignment = OrganizationFunctionAssignmentRef(
                id=assignment_id,
                tenant_id=tenant_id,
                identity_id=identity_id,
                account_id=account_id,
                function_id=function_id,
                organization_unit_id=organization_unit_id,
                applies_to_subunits=True,
                source="direct",
            )

            class FakeIdmDirectory:
                def get_organization_function_assignment(self, requested_assignment_id: str):
                    return assignment if requested_assignment_id == assignment_id else None

                def organization_function_assignments_for_identity(
                    self,
                    requested_identity_id: str,
                    *,
                    tenant_id: str | None = None,
                    effective_at=None,
                ):
                    del effective_at
                    if requested_identity_id == identity_id and tenant_id in (None, assignment.tenant_id):
                        return (assignment,)
                    return ()

                def organization_function_assignments_for_account(
                    self,
                    requested_account_id: str,
                    *,
                    tenant_id: str | None = None,
                    effective_at=None,
                ):
                    del effective_at
                    if requested_account_id == account_id and tenant_id in (None, assignment.tenant_id):
                        return (assignment,)
                    return ()

                def organization_function_assignments_for_identities(
                    self,
                    identity_ids,
                    *,
                    tenant_id: str | None = None,
                    effective_at=None,
                ):
                    return {
                        item_id: self.organization_function_assignments_for_identity(
                            item_id,
                            tenant_id=tenant_id,
                            effective_at=effective_at,
                        )
                        for item_id in identity_ids
                    }

                def organization_function_assignments_for_accounts(
                    self,
                    account_ids,
                    *,
                    tenant_id: str | None = None,
                    effective_at=None,
                ):
                    return {
                        item_id: self.organization_function_assignments_for_account(
                            item_id,
                            tenant_id=tenant_id,
                            effective_at=effective_at,
                        )
                        for item_id in account_ids
                    }

            class FakeOrganizationDirectory(CoreOrganizationDirectory):
                def get_organization_unit(self, requested_organization_unit_id: str):
                    if requested_organization_unit_id != organization_unit_id:
                        return None
                    return CoreOrganizationUnitRef(
                        id=organization_unit_id,
                        tenant_id=tenant_id,
                        slug="registry",
                        name="Registry Office",
                    )

                def organization_units_for_tenant(self, requested_tenant_id: str):
                    if requested_tenant_id != tenant_id:
                        return ()
                    unit = self.get_organization_unit(organization_unit_id)
                    return (unit,) if unit is not None else ()

                def get_function(self, requested_function_id: str):
                    if requested_function_id != function_id:
                        return None
                    return CoreOrganizationFunctionRef(
                        id=function_id,
                        tenant_id=tenant_id,
                        organization_unit_id=organization_unit_id,
                        slug="registry-clerk",
                        name="Registry Clerk",
                    )

                def functions_for_organization_unit(self, requested_organization_unit_id: str, *, include_subunits: bool = False):
                    del include_subunits
                    if requested_organization_unit_id != organization_unit_id:
                        return ()
                    function = self.get_function(function_id)
                    return (function,) if function is not None else ()

            idm_directory = FakeIdmDirectory()
            organization_directory = FakeOrganizationDirectory()

            class FakeFileAccessProvider:
                def explain_resource_provenance(
                    self,
                    session: object,
                    principal: PrincipalRef,
                    *,
                    resource_type: str,
                    resource_id: str,
                    action: str,
                ):
                    del session, principal, action
                    if resource_type != "file":
                        return ()
                    return (
                        AccessDecisionProvenance(
                            kind="resource",
                            id=resource_id,
                            label="Registry.pdf",
                            tenant_id=tenant_id,
                            source="files.file",
                            details={"resource_type": "file"},
                        ),
                    )

            resource_provider = FakeFileAccessProvider()

            with temporary_database(f"sqlite:///{root / 'test.db'}") as database:
                create_scope_tables(database.engine)
                Base.metadata.create_all(bind=database.engine)
                with database.session() as session:
                    tenant = Tenant(id=tenant_id, slug="access-explanation", name="Access Explanation", settings={})
                    account = Account(
                        id=account_id,
                        email="explanation@example.test",
                        normalized_email="explanation@example.test",
                        display_name="Access Explanation User",
                    )
                    user = User(
                        id=user_id,
                        tenant_id=tenant_id,
                        account_id=account_id,
                        email=account.email,
                        display_name=account.display_name,
                        settings={},
                        mail_profile_policy={},
                    )
                    role = Role(
                        id=role_id,
                        tenant_id=tenant_id,
                        slug="file-reader",
                        name="File Reader",
                        permissions=["files:file:read"],
                    )
                    mapping = ExternalFunctionRoleAssignment(
                        tenant_id=tenant_id,
                        source_module="organizations",
                        function_id=function_id,
                        role_id=role_id,
                    )
                    session.add_all([tenant, account, user, role, mapping])
                    session.commit()

                    from govoplan_access.backend.explanation import SqlAccessExplanationService, build_user_access_explanation

                    explanation = build_user_access_explanation(
                        session,
                        user,
                        idm_directory=idm_directory,
                        organization_directory=organization_directory,
                    )

                    self.assertEqual(1, len(explanation.function_facts))
                    self.assertEqual("Registry Clerk", explanation.function_facts[0].function_name)
                    self.assertEqual((role_id,), explanation.function_facts[0].role_ids)
                    source = next(item for item in explanation.role_sources if item.source_type == "idm_function_role")
                    self.assertEqual("Registry Clerk", source.function_name)
                    self.assertEqual("File Reader", source.role_name)
                    scope = next(item for item in explanation.scopes if item.scope == "files:file:read")
                    self.assertEqual(["idm_function_role"], [item.source_type for item in scope.sources])

                service = SqlAccessExplanationService(
                    idm_directory=idm_directory,
                    organization_directory=organization_directory,
                    resource_explanation_providers=(resource_provider,),
                )
                provenance = service.explain_scope_provenance(
                    PrincipalRef(
                        account_id=account_id,
                        membership_id=user_id,
                        tenant_id=tenant_id,
                        identity_id=identity_id,
                        scopes=frozenset({"files:file:read"}),
                    ),
                    "files:file:read",
                )
                self.assertTrue(any(item.kind == "function" and item.label == "Registry Clerk" for item in provenance))
                self.assertTrue(any(item.kind == "role" and item.label == "File Reader" for item in provenance))
                resource_provenance = service.explain_resource_provenance(
                    PrincipalRef(
                        account_id=account_id,
                        membership_id=user_id,
                        tenant_id=tenant_id,
                        identity_id=identity_id,
                        scopes=frozenset({"files:file:read"}),
                    ),
                    resource_type="file",
                    resource_id="file-access-explanation",
                    action="files:file:read",
                )
                self.assertTrue(any(item.kind == "resource" and item.id == "file-access-explanation" for item in resource_provenance))
                self.assertTrue(any(item.kind == "right" and item.id == "files:file:read" for item in resource_provenance))

                registry = PlatformRegistry()
                registry.register(
                    ModuleManifest(
                        id="access",
                        name="Access",
                        version="test",
                        capability_factories={CAPABILITY_ACCESS_EXPLANATION: lambda context: service},
                    )
                )
                registry.register(
                    ModuleManifest(
                        id="idm",
                        name="IDM",
                        version="test",
                        capability_factories={CAPABILITY_IDM_DIRECTORY: lambda context: idm_directory},
                    )
                )
                registry.register(
                    ModuleManifest(
                        id="organizations",
                        name="Organizations",
                        version="test",
                        capability_factories={CAPABILITY_ORGANIZATION_DIRECTORY: lambda context: organization_directory},
                    )
                )
                context = ModuleContext(registry=registry, settings=object())
                registry.configure_capability_context(context)
                configure_runtime(context)

                from govoplan_access.backend.api.v1.routes import router as admin_router

                app = FastAPI()
                app.include_router(admin_router)

                def principal_override() -> ApiPrincipal:
                    with database.session() as session:
                        account = session.get(Account, account_id)
                        user = session.get(User, user_id)
                        assert account is not None
                        assert user is not None
                        return ApiPrincipal(
                            principal=PrincipalRef(
                                account_id=account_id,
                                membership_id=user_id,
                                tenant_id=tenant_id,
                                scopes=frozenset({"admin:users:read"}),
                            ),
                            account=account,
                            user=user,
                        )

                app.dependency_overrides[get_api_principal] = principal_override
                with TestClient(app) as client:
                    response = client.get(f"/admin/users/{user_id}/access-explanation")
                self.assertEqual(200, response.status_code, response.text)
                payload = response.json()
                self.assertIn("files:file:read", payload["user"]["effective_scopes"])
                self.assertEqual("idm_function_role", payload["role_sources"][0]["source_type"])
                self.assertEqual("Registry Clerk", payload["function_facts"][0]["function_name"])
                resource_response = client.get(
                    "/admin/access/resource-explanation",
                    params={
                        "user_id": user_id,
                        "resource_type": "file",
                        "resource_id": "file-access-explanation",
                        "action": "files:file:read",
                    },
                )
                self.assertEqual(200, resource_response.status_code, resource_response.text)
                resource_payload = resource_response.json()
                self.assertTrue(any(item["kind"] == "resource" and item["id"] == "file-access-explanation" for item in resource_payload["provenance"]))
        finally:
            clear_runtime()
            shutil.rmtree(root, ignore_errors=True)

    def test_api_principal_dependency_uses_auth_provider_capability(self) -> None:
        root = Path(tempfile.mkdtemp(prefix="govoplan-access-contract-"))
        try:
            account_id = "account-capability"
            tenant_id = "tenant-capability"
            user_id = "user-capability"

            class CapabilityPrincipalResolver:
                def resolve_request(self, request: object, *, session: object | None = None) -> PrincipalRef:
                    del request, session
                    return PrincipalRef(
                        account_id=account_id,
                        membership_id=user_id,
                        tenant_id=tenant_id,
                        scopes=frozenset(),
                    )

            class CapabilityPermissionEvaluator:
                def scopes_grant(self, scopes, required_scope: str) -> bool:
                    del scopes
                    return required_scope == "demo:scope:read"

                def has_scope(self, principal: PrincipalRef, required_scope: str) -> bool:
                    del principal
                    return required_scope == "demo:scope:read"

                def explain_scope(self, principal: PrincipalRef, required_scope: str) -> AccessDecision:
                    return AccessDecision(allowed=self.has_scope(principal, required_scope), requirements=(required_scope,))

            class CapabilityApiPrincipalProvider:
                def resolve_api_principal(
                    self,
                    request,
                    session,
                    *,
                    authorization=None,
                    x_api_key=None,
                ):
                    del authorization, x_api_key
                    principal_ref = CapabilityPrincipalResolver().resolve_request(
                        request,
                        session=session,
                    )
                    account = session.get(Account, principal_ref.account_id)
                    user = session.get(User, principal_ref.membership_id)
                    assert account is not None
                    assert user is not None
                    return ApiPrincipal(
                        principal=principal_ref,
                        account=account,
                        user=user,
                        permission_evaluator=CapabilityPermissionEvaluator(),
                    )

            router = APIRouter()

            @router.get("/capability-principal")
            def capability_principal(principal: ApiPrincipal = Depends(get_api_principal)) -> dict[str, object]:
                return {
                    "account_id": principal.account_id,
                    "tenant_id": principal.tenant_id,
                    "user_id": principal.user.id,
                    "has_demo_scope": principal.has("demo:scope:read"),
                }

            def access_manifest() -> ModuleManifest:
                return ModuleManifest(
                    id=ACCESS_MODULE_ID,
                    name="Access",
                    version="test",
                    capability_factories={
                        CAPABILITY_ACCESS_PRINCIPAL_RESOLVER: lambda context: CapabilityPrincipalResolver(),
                        CAPABILITY_ACCESS_PERMISSION_EVALUATOR: lambda context: CapabilityPermissionEvaluator(),
                        CAPABILITY_AUTH_API_PRINCIPAL_PROVIDER: lambda context: CapabilityApiPrincipalProvider(),
                    },
                )

            config = GovoplanServerConfig(
                title="access contract test",
                version="test",
                settings=SimpleNamespace(database_url=f"sqlite:///{root / 'test.db'}"),
                enabled_modules=(ACCESS_MODULE_ID,),
                manifest_factories=(access_manifest,),
                base_routers=(router,),
                post_module_routers=(),
                extra_routers=(),
                lifespan=None,
                app_configurators=(),
            )
            with temporary_database(f"sqlite:///{root / 'test.db'}") as database:
                from govoplan_core.server.app import create_app

                app = create_app(config)

                create_scope_tables(database.engine)
                Base.metadata.create_all(bind=database.engine)
                with database.session() as session:
                    tenant = Tenant(id=tenant_id, slug="capability", name="Capability Tenant", settings={})
                    account = Account(
                        id=account_id,
                        email="capability@example.test",
                        normalized_email="capability@example.test",
                        display_name="Capability User",
                    )
                    user = User(
                        id=user_id,
                        tenant_id=tenant_id,
                        account_id=account_id,
                        email=account.email,
                        display_name=account.display_name,
                        settings={},
                        mail_profile_policy={},
                    )
                    session.add_all([tenant, account, user])
                    session.commit()

                with TestClient(app) as client:
                    response = client.get("/api/v1/capability-principal")

            self.assertEqual(response.status_code, 200, response.text)
            self.assertEqual(
                {
                    "account_id": account_id,
                    "tenant_id": tenant_id,
                    "user_id": user_id,
                    "has_demo_scope": True,
                },
                response.json(),
            )
        finally:
            shutil.rmtree(root, ignore_errors=True)


if __name__ == "__main__":
    unittest.main()
