from __future__ import annotations

import base64
import hashlib
from dataclasses import dataclass
from pathlib import Path
from secrets import token_hex
from typing import TYPE_CHECKING, Any, Iterator, Mapping
from urllib.parse import parse_qs, urlparse

import pytest
from smtpdfix.certs import Cert, _generate_certs

from .._helpers import (
    _AppInfo,
    _ExistingSpan,
    _httpx_client,
    _insert_spans,
    _OIDCServer,
    _server,
)

if TYPE_CHECKING:
    from phoenix.server.redaction import Redactor
    from tests.integration._mock_ldap_server import _LDAPServer


@pytest.fixture(scope="package")
def _env_oauth2_standard(
    _oidc_server_standard: _OIDCServer,
) -> dict[str, str]:
    """Configure standard OAuth2/OIDC environment variables (confidential client)."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_standard}_CLIENT_ID".upper(): _oidc_server_standard.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_standard}_CLIENT_SECRET".upper(): _oidc_server_standard.client_secret,
        f"PHOENIX_OAUTH2_{_oidc_server_standard}_OIDC_CONFIG_URL".upper(): f"{_oidc_server_standard.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_standard}_NO_SIGN_UP_CLIENT_ID".upper(): _oidc_server_standard.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_standard}_NO_SIGN_UP_CLIENT_SECRET".upper(): _oidc_server_standard.client_secret,
        f"PHOENIX_OAUTH2_{_oidc_server_standard}_NO_SIGN_UP_OIDC_CONFIG_URL".upper(): f"{_oidc_server_standard.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_standard}_NO_SIGN_UP_ALLOW_SIGN_UP".upper(): "false",
    }


@pytest.fixture(scope="package")
def _env_oauth2_pkce_public(
    _oidc_server_pkce_public: _OIDCServer,
) -> dict[str, str]:
    """Configure PKCE OAuth2 environment variables for public client (no client_secret)."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_public}_CLIENT_ID".upper(): _oidc_server_pkce_public.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_public}_OIDC_CONFIG_URL".upper(): f"{_oidc_server_pkce_public.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_public}_USE_PKCE".upper(): "true",
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_public}_TOKEN_ENDPOINT_AUTH_METHOD".upper(): "none",
    }


@pytest.fixture(scope="package")
def _env_oauth2_pkce_confidential(
    _oidc_server_pkce_confidential: _OIDCServer,
) -> dict[str, str]:
    """Configure PKCE OAuth2 environment variables for confidential client (defense-in-depth)."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_confidential}_CLIENT_ID".upper(): _oidc_server_pkce_confidential.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_confidential}_CLIENT_SECRET".upper(): _oidc_server_pkce_confidential.client_secret,
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_confidential}_OIDC_CONFIG_URL".upper(): f"{_oidc_server_pkce_confidential.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_confidential}_USE_PKCE".upper(): "true",
    }


@pytest.fixture(scope="package")
def _env_oauth2_pkce_groups_granted(
    _oidc_server_pkce_with_groups: _OIDCServer,
) -> dict[str, str]:
    """Configure PKCE OAuth2 with group access control - user HAS matching group."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_with_groups}_GRANTED_CLIENT_ID".upper(): _oidc_server_pkce_with_groups.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_with_groups}_GRANTED_OIDC_CONFIG_URL".upper(): f"{_oidc_server_pkce_with_groups.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_with_groups}_GRANTED_USE_PKCE".upper(): "true",
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_with_groups}_GRANTED_TOKEN_ENDPOINT_AUTH_METHOD".upper(): "none",
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_with_groups}_GRANTED_GROUPS_ATTRIBUTE_PATH".upper(): "groups",
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_with_groups}_GRANTED_ALLOWED_GROUPS".upper(): "engineering,admin",
    }


@pytest.fixture(scope="package")
def _env_oauth2_pkce_groups_denied(
    _oidc_server_pkce_with_groups: _OIDCServer,
) -> dict[str, str]:
    """Configure PKCE OAuth2 with group access control - user does NOT have matching group."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_with_groups}_DENIED_CLIENT_ID".upper(): _oidc_server_pkce_with_groups.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_with_groups}_DENIED_OIDC_CONFIG_URL".upper(): f"{_oidc_server_pkce_with_groups.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_with_groups}_DENIED_USE_PKCE".upper(): "true",
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_with_groups}_DENIED_TOKEN_ENDPOINT_AUTH_METHOD".upper(): "none",
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_with_groups}_DENIED_GROUPS_ATTRIBUTE_PATH".upper(): "groups",
        f"PHOENIX_OAUTH2_{_oidc_server_pkce_with_groups}_DENIED_ALLOWED_GROUPS".upper(): "admin,sales",
    }


@pytest.fixture(scope="package")
def _env_oauth2_standard_groups_granted(
    _oidc_server_standard_with_groups: _OIDCServer,
) -> dict[str, str]:
    """Configure standard OIDC with group access control - user HAS matching group."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_standard_with_groups}_GRANTED_CLIENT_ID".upper(): _oidc_server_standard_with_groups.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_standard_with_groups}_GRANTED_CLIENT_SECRET".upper(): _oidc_server_standard_with_groups.client_secret,
        f"PHOENIX_OAUTH2_{_oidc_server_standard_with_groups}_GRANTED_OIDC_CONFIG_URL".upper(): f"{_oidc_server_standard_with_groups.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_standard_with_groups}_GRANTED_GROUPS_ATTRIBUTE_PATH".upper(): "groups",
        f"PHOENIX_OAUTH2_{_oidc_server_standard_with_groups}_GRANTED_ALLOWED_GROUPS".upper(): "engineering,admin",
    }


@pytest.fixture(scope="package")
def _env_oauth2_standard_groups_denied(
    _oidc_server_standard_with_groups: _OIDCServer,
) -> dict[str, str]:
    """Configure standard OIDC with group access control - user does NOT have matching group."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_standard_with_groups}_DENIED_CLIENT_ID".upper(): _oidc_server_standard_with_groups.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_standard_with_groups}_DENIED_CLIENT_SECRET".upper(): _oidc_server_standard_with_groups.client_secret,
        f"PHOENIX_OAUTH2_{_oidc_server_standard_with_groups}_DENIED_OIDC_CONFIG_URL".upper(): f"{_oidc_server_standard_with_groups.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_standard_with_groups}_DENIED_GROUPS_ATTRIBUTE_PATH".upper(): "groups",
        f"PHOENIX_OAUTH2_{_oidc_server_standard_with_groups}_DENIED_ALLOWED_GROUPS".upper(): "admin,sales",
    }


@pytest.fixture(scope="package")
def _env_oauth2_role_admin(
    _oidc_server_with_role_admin: _OIDCServer,
) -> dict[str, str]:
    """Configure OAuth2 with role mapping - Owner → ADMIN."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_admin}_ADMIN_CLIENT_ID".upper(): _oidc_server_with_role_admin.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_admin}_ADMIN_CLIENT_SECRET".upper(): _oidc_server_with_role_admin.client_secret,
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_admin}_ADMIN_OIDC_CONFIG_URL".upper(): f"{_oidc_server_with_role_admin.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_admin}_ADMIN_ROLE_ATTRIBUTE_PATH".upper(): "role",
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_admin}_ADMIN_ROLE_MAPPING".upper(): "Owner:ADMIN,Developer:MEMBER,Reader:VIEWER",
    }


@pytest.fixture(scope="package")
def _env_oauth2_dynamic(
    _oidc_server_dynamic: _OIDCServer,
) -> dict[str, str]:
    """Configure OAuth2 with dynamic attribute changes (for testing IDP updates between logins)."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_dynamic}_DYNAMIC_CLIENT_ID".upper(): _oidc_server_dynamic.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_dynamic}_DYNAMIC_CLIENT_SECRET".upper(): _oidc_server_dynamic.client_secret,
        f"PHOENIX_OAUTH2_{_oidc_server_dynamic}_DYNAMIC_OIDC_CONFIG_URL".upper(): f"{_oidc_server_dynamic.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_dynamic}_DYNAMIC_ROLE_ATTRIBUTE_PATH".upper(): "role",
        f"PHOENIX_OAUTH2_{_oidc_server_dynamic}_DYNAMIC_ROLE_MAPPING".upper(): "Owner:ADMIN,Developer:MEMBER,Reader:VIEWER",
    }


@pytest.fixture(scope="package")
def _env_oauth2_role_member(
    _oidc_server_with_role_member: _OIDCServer,
) -> dict[str, str]:
    """Configure OAuth2 with role mapping - Developer → MEMBER."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_member}_MEMBER_CLIENT_ID".upper(): _oidc_server_with_role_member.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_member}_MEMBER_CLIENT_SECRET".upper(): _oidc_server_with_role_member.client_secret,
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_member}_MEMBER_OIDC_CONFIG_URL".upper(): f"{_oidc_server_with_role_member.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_member}_MEMBER_ROLE_ATTRIBUTE_PATH".upper(): "role",
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_member}_MEMBER_ROLE_MAPPING".upper(): "Owner:ADMIN,Developer:MEMBER,Reader:VIEWER",
    }


@pytest.fixture(scope="package")
def _env_oauth2_role_viewer(
    _oidc_server_with_role_viewer: _OIDCServer,
) -> dict[str, str]:
    """Configure OAuth2 with role mapping - Reader → VIEWER."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_viewer}_VIEWER_CLIENT_ID".upper(): _oidc_server_with_role_viewer.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_viewer}_VIEWER_CLIENT_SECRET".upper(): _oidc_server_with_role_viewer.client_secret,
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_viewer}_VIEWER_OIDC_CONFIG_URL".upper(): f"{_oidc_server_with_role_viewer.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_viewer}_VIEWER_ROLE_ATTRIBUTE_PATH".upper(): "role",
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_viewer}_VIEWER_ROLE_MAPPING".upper(): "Owner:ADMIN,Developer:MEMBER,Reader:VIEWER",
    }


@pytest.fixture(scope="package")
def _env_oauth2_role_system(
    _oidc_server_with_role_system: _OIDCServer,
) -> dict[str, str]:
    """Configure OAuth2 to test SYSTEM role blocking (no explicit mapping needed - raw value test)."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_system}_SYSTEM_CLIENT_ID".upper(): _oidc_server_with_role_system.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_system}_SYSTEM_CLIENT_SECRET".upper(): _oidc_server_with_role_system.client_secret,
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_system}_SYSTEM_OIDC_CONFIG_URL".upper(): f"{_oidc_server_with_role_system.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_with_role_system}_SYSTEM_ROLE_ATTRIBUTE_PATH".upper(): "role",
        # No role_mapping - tests raw value handling
    }


@pytest.fixture(scope="package")
def _env_oauth2_role_invalid_non_strict(
    _oidc_server_with_invalid_role: _OIDCServer,
) -> dict[str, str]:
    """Configure OAuth2 with role mapping - Invalid role defaults to VIEWER (non-strict)."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_with_invalid_role}_INVALID_CLIENT_ID".upper(): _oidc_server_with_invalid_role.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_with_invalid_role}_INVALID_CLIENT_SECRET".upper(): _oidc_server_with_invalid_role.client_secret,
        f"PHOENIX_OAUTH2_{_oidc_server_with_invalid_role}_INVALID_OIDC_CONFIG_URL".upper(): f"{_oidc_server_with_invalid_role.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_with_invalid_role}_INVALID_ROLE_ATTRIBUTE_PATH".upper(): "role",
        f"PHOENIX_OAUTH2_{_oidc_server_with_invalid_role}_INVALID_ROLE_MAPPING".upper(): "Owner:ADMIN,Developer:MEMBER,Reader:VIEWER",
        f"PHOENIX_OAUTH2_{_oidc_server_with_invalid_role}_INVALID_ROLE_ATTRIBUTE_STRICT".upper(): "false",
    }


@pytest.fixture(scope="package")
def _env_oauth2_role_invalid_strict(
    _oidc_server_with_invalid_role: _OIDCServer,
) -> dict[str, str]:
    """Configure OAuth2 with role mapping - Invalid role denies access (strict mode)."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_with_invalid_role}_STRICT_CLIENT_ID".upper(): _oidc_server_with_invalid_role.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_with_invalid_role}_STRICT_CLIENT_SECRET".upper(): _oidc_server_with_invalid_role.client_secret,
        f"PHOENIX_OAUTH2_{_oidc_server_with_invalid_role}_STRICT_OIDC_CONFIG_URL".upper(): f"{_oidc_server_with_invalid_role.base_url}/.well-known/openid-configuration",
        f"PHOENIX_OAUTH2_{_oidc_server_with_invalid_role}_STRICT_ROLE_ATTRIBUTE_PATH".upper(): "role",
        f"PHOENIX_OAUTH2_{_oidc_server_with_invalid_role}_STRICT_ROLE_MAPPING".upper(): "Owner:ADMIN,Developer:MEMBER,Reader:VIEWER",
        f"PHOENIX_OAUTH2_{_oidc_server_with_invalid_role}_STRICT_ROLE_ATTRIBUTE_STRICT".upper(): "true",
    }


@pytest.fixture(scope="package")
def _env_oauth2_role_missing_defaults_viewer(
    _oidc_server_without_role: _OIDCServer,
) -> dict[str, str]:
    """Configure OAuth2 without role mapping - defaults to VIEWER."""
    return {
        f"PHOENIX_OAUTH2_{_oidc_server_without_role}_DEFAULT_CLIENT_ID".upper(): _oidc_server_without_role.client_id,
        f"PHOENIX_OAUTH2_{_oidc_server_without_role}_DEFAULT_CLIENT_SECRET".upper(): _oidc_server_without_role.client_secret,
        f"PHOENIX_OAUTH2_{_oidc_server_without_role}_DEFAULT_OIDC_CONFIG_URL".upper(): f"{_oidc_server_without_role.base_url}/.well-known/openid-configuration",
    }


@pytest.fixture(scope="package")
def _env_oauth2(
    _env_oauth2_standard: dict[str, str],
    _env_oauth2_pkce_public: dict[str, str],
    _env_oauth2_pkce_confidential: dict[str, str],
    _env_oauth2_pkce_groups_granted: dict[str, str],
    _env_oauth2_pkce_groups_denied: dict[str, str],
    _env_oauth2_standard_groups_granted: dict[str, str],
    _env_oauth2_standard_groups_denied: dict[str, str],
    _env_oauth2_dynamic: dict[str, str],
    _env_oauth2_role_admin: dict[str, str],
    _env_oauth2_role_member: dict[str, str],
    _env_oauth2_role_viewer: dict[str, str],
    _env_oauth2_role_system: dict[str, str],
    _env_oauth2_role_invalid_non_strict: dict[str, str],
    _env_oauth2_role_invalid_strict: dict[str, str],
    _env_oauth2_role_missing_defaults_viewer: dict[str, str],
) -> dict[str, str]:
    """Combine all OAuth2 environment configurations for testing."""
    return {
        **_env_oauth2_standard,
        **_env_oauth2_pkce_public,
        **_env_oauth2_pkce_confidential,
        **_env_oauth2_pkce_groups_granted,
        **_env_oauth2_pkce_groups_denied,
        **_env_oauth2_standard_groups_granted,
        **_env_oauth2_standard_groups_denied,
        **_env_oauth2_dynamic,
        **_env_oauth2_role_admin,
        **_env_oauth2_role_member,
        **_env_oauth2_role_viewer,
        **_env_oauth2_role_system,
        **_env_oauth2_role_invalid_non_strict,
        **_env_oauth2_role_invalid_strict,
        **_env_oauth2_role_missing_defaults_viewer,
    }


@pytest.fixture(scope="package")
def _env_ldap(_ldap_server: _LDAPServer) -> dict[str, str]:
    """Configure LDAP environment variables for testing with mock LDAP server."""
    return {
        "PHOENIX_LDAP_HOST": _ldap_server.host,
        "PHOENIX_LDAP_PORT": str(_ldap_server.port),
        "PHOENIX_LDAP_TLS_MODE": "none",  # Disable TLS for mock testing
        "PHOENIX_LDAP_BIND_DN": _ldap_server.bind_dn,
        "PHOENIX_LDAP_BIND_PASSWORD": _ldap_server.bind_password,
        "PHOENIX_LDAP_USER_SEARCH_BASE_DNS": f'["{_ldap_server.user_search_base}"]',
        "PHOENIX_LDAP_USER_SEARCH_FILTER": "(uid=%s)",
        "PHOENIX_LDAP_ATTR_EMAIL": "mail",
        "PHOENIX_LDAP_ATTR_DISPLAY_NAME": "displayName",
        "PHOENIX_LDAP_ATTR_MEMBER_OF": "memberOf",
        "PHOENIX_LDAP_GROUP_ROLE_MAPPINGS": (
            '[{"group_dn": "cn=admins,ou=groups,dc=example,dc=com", "role": "ADMIN"}, '
            '{"group_dn": "cn=members,ou=groups,dc=example,dc=com", "role": "MEMBER"}, '
            '{"group_dn": "*", "role": "VIEWER"}]'
        ),
        # Default: allow_sign_up=true (users auto-created on first login)
    }


@pytest.fixture(scope="package")
def _env_ldap_no_sign_up(_env_ldap: Mapping[str, str]) -> dict[str, str]:
    """Configure LDAP with allow_sign_up=false (admin must pre-create users)."""
    return {
        **_env_ldap,
        "PHOENIX_LDAP_ALLOW_SIGN_UP": "false",
    }


@pytest.fixture(scope="package")
def _env_ldap_unique_id(_env_ldap: Mapping[str, str]) -> dict[str, str]:
    """Configure LDAP with unique_id attribute (enterprise mode).

    Uses entryUUID as the unique identifier. This enables:
    - Identity preserved across email changes
    - Identity preserved across DN changes
    - Compliance scenarios requiring stable user identifiers
    """
    return {
        **_env_ldap,
        "PHOENIX_LDAP_ATTR_UNIQUE_ID": "entryUUID",
    }


@pytest.fixture(scope="package")
def _env_ldap_no_email(_ldap_server: _LDAPServer) -> dict[str, str]:
    """Configure LDAP with no email attribute (null email marker mode).

    This fixture tests the no-email mode where LDAP directories don't have
    email attributes. Phoenix generates null email markers using the unique_id.

    Requirements for no-email mode:
    - PHOENIX_LDAP_ATTR_EMAIL="" (empty string)
    - PHOENIX_LDAP_ATTR_UNIQUE_ID required (users identified by unique_id)
    - PHOENIX_LDAP_ALLOW_SIGN_UP=true (users must be auto-provisioned)
    """
    return {
        "PHOENIX_LDAP_HOST": _ldap_server.host,
        "PHOENIX_LDAP_PORT": str(_ldap_server.port),
        "PHOENIX_LDAP_TLS_MODE": "none",  # Disable TLS for mock testing
        "PHOENIX_LDAP_BIND_DN": _ldap_server.bind_dn,
        "PHOENIX_LDAP_BIND_PASSWORD": _ldap_server.bind_password,
        "PHOENIX_LDAP_USER_SEARCH_BASE_DNS": f'["{_ldap_server.user_search_base}"]',
        "PHOENIX_LDAP_USER_SEARCH_FILTER": "(uid=%s)",
        "PHOENIX_LDAP_ATTR_EMAIL": "",  # Empty = no-email mode
        "PHOENIX_LDAP_ATTR_DISPLAY_NAME": "displayName",
        "PHOENIX_LDAP_ATTR_MEMBER_OF": "memberOf",
        "PHOENIX_LDAP_ATTR_UNIQUE_ID": "entryUUID",  # Required for no-email mode
        "PHOENIX_LDAP_GROUP_ROLE_MAPPINGS": (
            '[{"group_dn": "cn=admins,ou=groups,dc=example,dc=com", "role": "ADMIN"}, '
            '{"group_dn": "cn=members,ou=groups,dc=example,dc=com", "role": "MEMBER"}, '
            '{"group_dn": "*", "role": "VIEWER"}]'
        ),
        "PHOENIX_LDAP_ALLOW_SIGN_UP": "true",  # Required for no-email mode
    }


@pytest.fixture(scope="package")
def _env_ldap_posix(_ldap_server: _LDAPServer) -> dict[str, str]:
    """Configure LDAP with group search instead of memberOf attribute.

    This fixture tests group lookup via search rather than reading the memberOf
    attribute from the user entry. Since GROUP_SEARCH_FILTER_USER_ATTR is not set,
    %s in the filter is replaced with the login username directly.

    Key differences from _env_ldap:
    - No ATTR_MEMBER_OF (relies on group search)
    - Adds GROUP_SEARCH_BASE_DNS and GROUP_SEARCH_FILTER

    Note: This does NOT test the GROUP_SEARCH_FILTER_USER_ATTR code path.
    See _env_ldap_posix_memberuid for that.
    """
    return {
        "PHOENIX_LDAP_HOST": _ldap_server.host,
        "PHOENIX_LDAP_PORT": str(_ldap_server.port),
        "PHOENIX_LDAP_TLS_MODE": "none",  # Disable TLS for mock testing
        "PHOENIX_LDAP_BIND_DN": _ldap_server.bind_dn,
        "PHOENIX_LDAP_BIND_PASSWORD": _ldap_server.bind_password,
        "PHOENIX_LDAP_USER_SEARCH_BASE_DNS": f'["{_ldap_server.user_search_base}"]',
        "PHOENIX_LDAP_USER_SEARCH_FILTER": "(uid=%s)",
        "PHOENIX_LDAP_ATTR_EMAIL": "mail",
        "PHOENIX_LDAP_ATTR_DISPLAY_NAME": "displayName",
        "PHOENIX_LDAP_ATTR_MEMBER_OF": "",
        "PHOENIX_LDAP_GROUP_SEARCH_BASE_DNS": f'["{_ldap_server.group_search_base}"]',
        "PHOENIX_LDAP_GROUP_SEARCH_FILTER": "(member=%s)",  # %s replaced with login username
        "PHOENIX_LDAP_GROUP_ROLE_MAPPINGS": (
            '[{"group_dn": "cn=admins,ou=groups,dc=example,dc=com", "role": "ADMIN"}, '
            '{"group_dn": "cn=members,ou=groups,dc=example,dc=com", "role": "MEMBER"}, '
            '{"group_dn": "*", "role": "VIEWER"}]'
        ),
        # Default: allow_sign_up=true (users auto-created on first login)
    }


@pytest.fixture(scope="package")
def _env_ldap_posix_memberuid(_ldap_server: _LDAPServer) -> dict[str, str]:
    """Configure LDAP with POSIX memberUid group search (RFC 2307 style).

    This fixture tests the POSIX group lookup method where:
    - Groups use memberUid attribute with username values (not DNs)
    - GROUP_SEARCH_FILTER_USER_ATTR specifies which user attribute to use

    Key differences from _env_ldap_posix:
    - Uses (memberUid=%s) filter instead of (member=%s)
    - Sets GROUP_SEARCH_FILTER_USER_ATTR=uid to read uid from user entry

    This tests the code path where Phoenix must fetch a specific attribute
    from the user entry for group filter substitution, which was previously
    buggy (the uid attribute wasn't being fetched).
    """
    return {
        "PHOENIX_LDAP_HOST": _ldap_server.host,
        "PHOENIX_LDAP_PORT": str(_ldap_server.port),
        "PHOENIX_LDAP_TLS_MODE": "none",
        "PHOENIX_LDAP_BIND_DN": _ldap_server.bind_dn,
        "PHOENIX_LDAP_BIND_PASSWORD": _ldap_server.bind_password,
        "PHOENIX_LDAP_USER_SEARCH_BASE_DNS": f'["{_ldap_server.user_search_base}"]',
        "PHOENIX_LDAP_USER_SEARCH_FILTER": "(uid=%s)",
        "PHOENIX_LDAP_ATTR_EMAIL": "mail",
        "PHOENIX_LDAP_ATTR_DISPLAY_NAME": "displayName",
        "PHOENIX_LDAP_ATTR_MEMBER_OF": "",
        "PHOENIX_LDAP_GROUP_SEARCH_BASE_DNS": f'["{_ldap_server.group_search_base}"]',
        "PHOENIX_LDAP_GROUP_SEARCH_FILTER": "(memberUid=%s)",  # %s replaced with uid attr
        "PHOENIX_LDAP_GROUP_SEARCH_FILTER_USER_ATTR": "uid",  # Read uid from user entry
        "PHOENIX_LDAP_GROUP_ROLE_MAPPINGS": (
            '[{"group_dn": "cn=admins,ou=groups,dc=example,dc=com", "role": "ADMIN"}, '
            '{"group_dn": "cn=members,ou=groups,dc=example,dc=com", "role": "MEMBER"}, '
            '{"group_dn": "*", "role": "VIEWER"}]'
        ),
    }


@pytest.fixture(scope="package")
def _env_agents() -> dict[str, str]:
    """Configure agents testing environment.

    The chat auth integration tests use the Anthropic built-in provider only
    to exercise route authentication, not provider credential validation, so
    provide a fake key to keep those requests on the authenticated code path.
    """
    return {
        "ANTHROPIC_API_KEY": "sk-fake-anthropic-key",
    }


@pytest.fixture(scope="package")
def _env(
    _env_auth: Mapping[str, str],
    _env_database: Mapping[str, str],
    _env_oauth2: Mapping[str, str],
    _env_ldap: Mapping[str, str],
    _env_ports: Mapping[str, str],
    _env_smtp: Mapping[str, str],
    _env_tls: Mapping[str, str],
    _env_agents: Mapping[str, str],
) -> dict[str, str]:
    """Combine all environment variable configurations for testing."""
    return {
        **_env_tls,
        **_env_ports,
        **_env_database,
        **_env_auth,
        **_env_smtp,
        **_env_oauth2,
        **_env_ldap,
        **_env_agents,
        # The OAuth2 consent decision endpoint requires an Origin header that
        # matches the server's public origin. Tests reach the server at
        # 127.0.0.1, so that hostname must be trusted in addition to the
        # localhost default; the CSRF validator compares hostnames only.
        "PHOENIX_CSRF_TRUSTED_ORIGINS": ",http://localhost,http://127.0.0.1,",
        # The /mcp mount is additive (no existing route or behavior changes when
        # it is on), so the whole auth package runs against it: test_mcp.py
        # exercises it directly, and every other test doubles as evidence that
        # the mount does not interfere with the rest of the surface.
        "PHOENIX_ENABLE_MCP_SERVER": "true",
        # Pin the group-gated progressive-disclosure surface for the shared
        # package app. Code mode is the global default, but it replaces the tool
        # surface wholesale, so the package app keeps the group-gated list;
        # _app_mcp_code_mode is the dedicated fixture that opts into code mode.
        "PHOENIX_ENABLE_MCP_CODE_MODE": "false",
    }


@pytest.fixture(scope="package")
def _app(
    _env: dict[str, str],
) -> Iterator[_AppInfo]:
    with _server(_AppInfo(_env)) as app:
        yield app


@pytest.fixture(scope="package")
def _redactor(_app: _AppInfo) -> "Redactor":
    """A Redactor keyed off the same PHOENIX_SECRET as the running server.

    Tests use this to redact values before sending and to un-redact server-emitted
    tokens — verifying symmetric round-trips end-to-end.
    """
    from pydantic import SecretStr

    from phoenix.server.redaction import Redactor

    return Redactor(secret=SecretStr(_app.env["PHOENIX_SECRET"]))


@pytest.fixture(scope="package")
def _env_ports_ldap_no_sign_up(
    _ports: Iterator[int],
) -> dict[str, str]:
    """Separate port allocation for LDAP no-sign-up app."""
    return {
        "PHOENIX_PORT": str(next(_ports)),
        "PHOENIX_GRPC_PORT": str(next(_ports)),
    }


@pytest.fixture(scope="package")
def _app_ldap_no_sign_up(
    _env_auth: Mapping[str, str],
    _env_database: Mapping[str, str],
    _env_oauth2: Mapping[str, str],
    _env_ldap_no_sign_up: Mapping[str, str],
    _env_ports_ldap_no_sign_up: Mapping[str, str],
    _env_smtp: Mapping[str, str],
    _env_tls: Mapping[str, str],
) -> Iterator[_AppInfo]:
    """App instance with LDAP allow_sign_up=false.

    Uses separate ports from _app_ldap to allow both apps to run concurrently.
    """
    env = {
        **_env_tls,
        **_env_ports_ldap_no_sign_up,
        **_env_database,
        **_env_auth,
        **_env_smtp,
        **_env_oauth2,
        **_env_ldap_no_sign_up,
    }
    with _server(_AppInfo(env)) as app:
        yield app


@pytest.fixture(scope="package")
def _env_ports_posix(
    _ports: Iterator[int],
) -> dict[str, str]:
    """Separate port allocation for POSIX LDAP app to avoid conflicts with _app_ldap."""
    return {
        "PHOENIX_PORT": str(next(_ports)),
        "PHOENIX_GRPC_PORT": str(next(_ports)),
    }


@pytest.fixture(scope="package")
def _env_ports_ldap_no_email(
    _ports: Iterator[int],
) -> dict[str, str]:
    """Separate port allocation for LDAP no-email app."""
    return {
        "PHOENIX_PORT": str(next(_ports)),
        "PHOENIX_GRPC_PORT": str(next(_ports)),
    }


@pytest.fixture(scope="package")
def _env_ports_ldap_unique_id(
    _ports: Iterator[int],
) -> dict[str, str]:
    """Separate port allocation for LDAP unique_id app."""
    return {
        "PHOENIX_PORT": str(next(_ports)),
        "PHOENIX_GRPC_PORT": str(next(_ports)),
    }


@pytest.fixture(scope="package")
def _app_ldap_posix(
    _env_auth: Mapping[str, str],
    _env_database: Mapping[str, str],
    _env_oauth2: Mapping[str, str],
    _env_ldap_posix: Mapping[str, str],
    _env_ports_posix: Mapping[str, str],
    _env_smtp: Mapping[str, str],
    _env_tls: Mapping[str, str],
) -> Iterator[_AppInfo]:
    """App instance with LDAP configured for POSIX group search (OpenLDAP).

    Uses separate ports from _app_ldap to allow both apps to run concurrently
    during integration tests.
    """
    env = {
        **_env_tls,
        **_env_ports_posix,
        **_env_database,
        **_env_auth,
        **_env_smtp,
        **_env_oauth2,
        **_env_ldap_posix,
    }
    with _server(_AppInfo(env)) as app:
        yield app


@pytest.fixture(scope="package")
def _env_ports_posix_memberuid(
    _ports: Iterator[int],
) -> dict[str, str]:
    """Separate port allocation for POSIX memberUid LDAP app."""
    return {
        "PHOENIX_PORT": str(next(_ports)),
        "PHOENIX_GRPC_PORT": str(next(_ports)),
    }


@pytest.fixture(scope="package")
def _app_ldap_posix_memberuid(
    _env_auth: Mapping[str, str],
    _env_database: Mapping[str, str],
    _env_oauth2: Mapping[str, str],
    _env_ldap_posix_memberuid: Mapping[str, str],
    _env_ports_posix_memberuid: Mapping[str, str],
    _env_smtp: Mapping[str, str],
    _env_tls: Mapping[str, str],
) -> Iterator[_AppInfo]:
    """App instance with LDAP configured for POSIX memberUid group search.

    Uses GROUP_SEARCH_FILTER_USER_ATTR=uid to test the code path where Phoenix
    must fetch the uid attribute from the user entry for group filter substitution.
    """
    env = {
        **_env_tls,
        **_env_ports_posix_memberuid,
        **_env_database,
        **_env_auth,
        **_env_smtp,
        **_env_oauth2,
        **_env_ldap_posix_memberuid,
    }
    with _server(_AppInfo(env)) as app:
        yield app


@pytest.fixture(scope="package")
def _app_ldap_unique_id(
    _env_auth: Mapping[str, str],
    _env_database: Mapping[str, str],
    _env_oauth2: Mapping[str, str],
    _env_ldap_unique_id: Mapping[str, str],
    _env_ports_ldap_unique_id: Mapping[str, str],
    _env_smtp: Mapping[str, str],
    _env_tls: Mapping[str, str],
) -> Iterator[_AppInfo]:
    """App instance with LDAP configured for unique_id identification (enterprise mode).

    Uses PHOENIX_LDAP_ATTR_UNIQUE_ID=entryUUID to identify users by a stable
    unique identifier rather than email. This enables identity preservation
    across email and DN changes.
    """
    env = {
        **_env_tls,
        **_env_ports_ldap_unique_id,
        **_env_database,
        **_env_auth,
        **_env_smtp,
        **_env_oauth2,
        **_env_ldap_unique_id,
    }
    with _server(_AppInfo(env)) as app:
        yield app


@pytest.fixture(scope="package")
def _app_ldap_no_email(
    _env_auth: Mapping[str, str],
    _env_database: Mapping[str, str],
    _env_oauth2: Mapping[str, str],
    _env_ldap_no_email: Mapping[str, str],
    _env_ports_ldap_no_email: Mapping[str, str],
    _env_smtp: Mapping[str, str],
    _env_tls: Mapping[str, str],
) -> Iterator[_AppInfo]:
    """App instance with LDAP configured for no-email mode (null email markers).

    Uses PHOENIX_LDAP_ATTR_EMAIL="" to enable no-email mode where users are
    identified by entryUUID instead of email. Phoenix generates null email
    markers for the database.
    """
    env = {
        **_env_tls,
        **_env_ports_ldap_no_email,
        **_env_database,
        **_env_auth,
        **_env_smtp,
        **_env_oauth2,
        **_env_ldap_no_email,
    }
    with _server(_AppInfo(env)) as app:
        yield app


@pytest.fixture(scope="package")
def _env_tls(
    _tls_certs_for_server: Cert,
    _tls_certs_for_client: Cert,
) -> dict[str, str]:
    """Configure TLS environment variables for testing.

    This fixture sets up the necessary environment variables for TLS configuration
    in the Phoenix server. It encrypts the server's private key with a random password
    and configures both server and client certificates for mutual TLS authentication.

    The fixture is automatically used in all tests within its scope and patches
    the environment variables temporarily during test execution.

    Args:
        _tls_certs_server: Server TLS certificates fixture
        _tls_certs_client: Client TLS certificates fixture

    Yields:
        None: The fixture yields control back to the test after setting up the environment
    """
    key_file_password = token_hex(16)
    key_file = _encrypt_private_key(_tls_certs_for_server.key[0], key_file_password)
    return {
        "PHOENIX_TLS_ENABLED": "true",
        "PHOENIX_TLS_CERT_FILE": str(_tls_certs_for_server.cert.resolve()),
        "PHOENIX_TLS_KEY_FILE": str(key_file.resolve()),
        "PHOENIX_TLS_KEY_FILE_PASSWORD": key_file_password,
        "PHOENIX_TLS_CA_FILE": str(_tls_certs_for_client.cert.resolve()),
        "PHOENIX_TLS_VERIFY_CLIENT": "true",
    }


@pytest.fixture(scope="package")
def _tls_certs_for_server(
    tmp_path_factory: pytest.TempPathFactory,
) -> Cert:
    """Fixture that provides TLS certificates in a temporary directory."""
    path = tmp_path_factory.mktemp(f"certs_for_server_{token_hex(8)}")
    return _generate_certs(path, separate_key=True)


@pytest.fixture(scope="package")
def _tls_certs_for_client(
    tmp_path_factory: pytest.TempPathFactory,
) -> Cert:
    """Fixture that provides TLS certificates in a temporary directory."""
    path = tmp_path_factory.mktemp(f"certs_for_client_{token_hex(8)}")
    return _generate_certs(path, separate_key=False)


@pytest.fixture(scope="package")
def _oidc_server_standard(
    _ports: Iterator[int],
) -> Iterator[_OIDCServer]:
    """Standard OAuth2/OIDC server (confidential client with client_secret)."""
    with _OIDCServer(port=next(_ports), use_pkce=False) as server:
        yield server


@pytest.fixture(scope="package")
def _oidc_server_pkce_public(
    _ports: Iterator[int],
) -> Iterator[_OIDCServer]:
    """PKCE-enabled OIDC server for public clients (no client_secret)."""
    with _OIDCServer(port=next(_ports), use_pkce=True) as server:
        yield server


@pytest.fixture(scope="package")
def _oidc_server_pkce_confidential(
    _ports: Iterator[int],
) -> Iterator[_OIDCServer]:
    """PKCE-enabled OIDC server for confidential clients (defense-in-depth)."""
    with _OIDCServer(port=next(_ports), use_pkce=True) as server:
        yield server


@pytest.fixture(scope="package")
def _oidc_server_pkce_with_groups(
    _ports: Iterator[int],
) -> Iterator[_OIDCServer]:
    """PKCE-enabled OIDC server with group claims for access control testing."""
    with _OIDCServer(
        port=next(_ports), use_pkce=True, groups=["engineering", "operations"]
    ) as server:
        yield server


@pytest.fixture(scope="package")
def _oidc_server_standard_with_groups(
    _ports: Iterator[int],
) -> Iterator[_OIDCServer]:
    """Standard OIDC server with group claims for access control testing."""
    with _OIDCServer(
        port=next(_ports), use_pkce=False, groups=["engineering", "operations"]
    ) as server:
        yield server


# Backward compatibility alias
@pytest.fixture(scope="package")
def _oidc_server(
    _oidc_server_standard: _OIDCServer,
) -> _OIDCServer:
    """Alias for backward compatibility with existing tests."""
    return _oidc_server_standard


@pytest.fixture(scope="package")
def _oidc_server_dynamic(
    _ports: Iterator[int],
) -> Iterator[_OIDCServer]:
    """
    Dynamic OIDC server for testing IDP attribute changes between logins.

    This flexible server supports dynamically changing role, email, and user
    identity using set_role(), set_email(), and set_user() to simulate
    IDP changes between login sessions.

    Starts with 'Developer' role (maps to MEMBER), but all attributes can be changed
    dynamically for testing scenarios where the IDP state changes between logins.

    Examples:
        # Test role changes for same user across logins
        server.set_user("user_123", "alice@example.com", num_logins=2)
        server.set_role("Developer", num_logins=1)  # First login: MEMBER
        email1, _, _ = await complete_flow(app, server)

        server.set_role("Owner", num_logins=1)  # Second login: ADMIN
        email2, _, _ = await complete_flow(app, server)
        assert email1 == email2  # Same user, different role!

        # Test email changes for same user
        server.set_user("user_123", "old@example.com", num_logins=2)
        email1, _, _ = await complete_flow(app, server)  # old@example.com

        server.set_email("new@example.com", num_logins=1)
        email2, _, _ = await complete_flow(app, server)  # new@example.com
    """
    with _OIDCServer(port=next(_ports), use_pkce=False, role="Developer") as server:
        yield server


@pytest.fixture(scope="package")
def _oidc_server_with_role_admin(
    _ports: Iterator[int],
) -> Iterator[_OIDCServer]:
    """OIDC server with role claim set to 'Owner' (maps to ADMIN)."""
    with _OIDCServer(port=next(_ports), use_pkce=False, role="Owner") as server:
        yield server


@pytest.fixture(scope="package")
def _oidc_server_with_role_member(
    _ports: Iterator[int],
) -> Iterator[_OIDCServer]:
    """OIDC server with role claim set to 'Developer' (maps to MEMBER)."""
    with _OIDCServer(port=next(_ports), use_pkce=False, role="Developer") as server:
        yield server


@pytest.fixture(scope="package")
def _oidc_server_with_role_viewer(
    _ports: Iterator[int],
) -> Iterator[_OIDCServer]:
    """OIDC server with role claim set to 'Reader' (maps to VIEWER)."""
    with _OIDCServer(port=next(_ports), use_pkce=False, role="Reader") as server:
        yield server


@pytest.fixture(scope="package")
def _oidc_server_with_role_system(
    _ports: Iterator[int],
) -> Iterator[_OIDCServer]:
    """OIDC server with role claim set to 'SYSTEM' (should be blocked)."""
    with _OIDCServer(port=next(_ports), use_pkce=False, role="SYSTEM") as server:
        yield server


@pytest.fixture(scope="package")
def _oidc_server_with_invalid_role(
    _ports: Iterator[int],
) -> Iterator[_OIDCServer]:
    """OIDC server with role claim set to an invalid/unmapped role."""
    with _OIDCServer(port=next(_ports), use_pkce=False, role="InvalidRole") as server:
        yield server


@pytest.fixture(scope="package")
def _oidc_server_without_role(
    _ports: Iterator[int],
) -> Iterator[_OIDCServer]:
    """OIDC server without role claim (role=None)."""
    with _OIDCServer(port=next(_ports), use_pkce=False, role=None) as server:
        yield server


def _encrypt_private_key(key_path: Path, password: str) -> Path:
    """
    Encrypt an existing private key file with a password and save to a new file.

    Args:
        key_path: Path to the private key file (PEM format)
        password: Password to encrypt the key with

    Returns:
        Path: Path to the new encrypted key file
    """
    from cryptography.hazmat.backends import default_backend
    from cryptography.hazmat.primitives import serialization
    from cryptography.hazmat.primitives.serialization import load_pem_private_key

    # Create path for encrypted file
    encrypted_path = key_path.with_name(f"{key_path.stem}_encrypted{key_path.suffix}")

    # Read the unencrypted private key
    with open(key_path, "rb") as f:
        private_key_data = f.read()

    # Load the private key (it's currently unencrypted)
    private_key = load_pem_private_key(
        private_key_data,
        password=None,  # The key is not encrypted yet
        backend=default_backend(),
    )

    # Convert password to bytes
    password_bytes = password.encode()

    # Encrypt the private key
    encrypted_pem = private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.TraditionalOpenSSL,  # Match the original format
        encryption_algorithm=serialization.BestAvailableEncryption(password_bytes),
    )

    # Write the encrypted key to the new file
    with open(encrypted_path, "wb") as f:
        f.write(encrypted_pem)

    return encrypted_path


@pytest.fixture(scope="package")
def _existing_spans(
    _app: _AppInfo,
) -> tuple[_ExistingSpan, ...]:
    return _insert_spans(_app, 10)


# =============================================================================
# LDAP Test Fixtures
# =============================================================================


@pytest.fixture(scope="package")
def _ldap_server(_ports: Iterator[int]) -> Iterator[_LDAPServer]:
    """Start mock LDAP server for integration tests.

    This fixture provides a lightweight, in-process LDAP server that implements
    minimal LDAP protocol operations (bind, search) needed for testing Phoenix's
    LDAP authentication flow.

    The server is similar to _oidc_server - it runs in a separate thread and
    listens on a dynamically allocated port.
    """
    from tests.integration._mock_ldap_server import _LDAPServer

    with _LDAPServer(port=next(_ports)) as ldap_server:
        yield ldap_server


# =============================================================================
# OAuth2 Authorization Server Test Fixtures
# =============================================================================


@pytest.fixture(scope="package")
def _oauth_public_client(_app: _AppInfo) -> Iterator[_OAuthPublicClient]:
    """Register a public client through dynamic client registration.

    Registration goes through the real /oauth2/register endpoint so these
    tests exercise the app purely over HTTP, with no assumptions about which
    database (or schema) the server is running on.
    """
    name = "OAuth public test client"
    redirect_uri = "http://127.0.0.1:8765/callback"
    response = _httpx_client(_app).post(
        "oauth2/register",
        json={"client_name": name, "redirect_uris": [redirect_uri]},
    )
    response.raise_for_status()
    registration = response.json()
    yield _OAuthPublicClient(
        client_id=registration["client_id"],
        name=name,
        redirect_uri=redirect_uri,
        app=_app,
    )


@dataclass(frozen=True)
class _OAuthPublicClient:
    client_id: str
    name: str
    redirect_uri: str
    app: _AppInfo
    code_verifier: str = ""
    code_challenge: str = ""
    #: RFC 8707 resource indicator sent on both the authorize and token requests
    #: when set — the shape MCP clients use (their `resource` is the MCP endpoint).
    resource: str | None = None

    def __post_init__(self) -> None:
        verifier = "a" + token_hex(48)
        challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).decode()
        object.__setattr__(self, "code_verifier", verifier)
        object.__setattr__(self, "code_challenge", challenge.rstrip("="))

    def authorization_params(self, *, state: str | None = None) -> dict[str, str]:
        params = {
            "response_type": "code",
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri,
            "code_challenge": self.code_challenge,
            "code_challenge_method": "S256",
            "state": state or f"state-{token_hex(16)}",
            "scope": "read write ignored",
        }
        if self.resource is not None:
            params["resource"] = self.resource
        return params

    def authorize(self, auth: Any, *, state: str | None = None) -> dict[str, str]:
        params = self.authorization_params(state=state)
        response = _httpx_client(self.app, auth).get(
            "oauth2/authorize",
            params=params,
            follow_redirects=False,
        )
        assert response.status_code == 302
        assert urlparse(response.headers["location"]).path.endswith("/oauth2/consent")
        return params

    def decide(self, auth: Any, params: dict[str, str], *, approved: bool = True) -> str:
        response = _httpx_client(
            self.app, auth, headers={"origin": _origin(self.app.base_url)}
        ).post(
            "oauth2/authorize/decision",
            json={**params, "approved": approved},
        )
        response.raise_for_status()
        redirect_to = response.json()["redirect_to"]
        assert isinstance(redirect_to, str)
        return redirect_to

    def exchange_code(self, redirect_to: str) -> dict[str, Any]:
        query = parse_qs(urlparse(redirect_to).query)
        code = query["code"][0]
        data = {
            "grant_type": "authorization_code",
            "code": code,
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri,
            "code_verifier": self.code_verifier,
        }
        if self.resource is not None:
            data["resource"] = self.resource
        response = _httpx_client(self.app).post(
            "oauth2/token",
            data=data,
        )
        response.raise_for_status()
        data = response.json()
        assert isinstance(data, dict)
        return data

    def complete_flow(self, auth: Any) -> dict[str, Any]:
        params = self.authorize(auth)
        redirect_to = self.decide(auth, params)
        return self.exchange_code(redirect_to)


def _active_grants(app: _AppInfo, user: Any) -> list[dict[str, Any]]:
    """The user's active (non-revoked) grants, observed through the GraphQL API.

    Revoked grants are filtered out by the server, so a grant's presence in
    this list doubles as the assertion that it has not been revoked.
    """
    response, _ = user.gql(
        app,
        "query { viewer { ... on User { oauth2Grants { id clientId scopes } } } }",
    )
    assert not response.get("errors")
    grants = response["data"]["viewer"]["oauth2Grants"]
    assert isinstance(grants, list)
    return grants


@pytest.fixture
def _app_dcr_rate_limited(
    _ports: Iterator[int],
    tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[_AppInfo]:
    port = next(_ports)
    env = _oauth2_app_env(
        port=port,
        grpc_port=next(_ports),
        database=str(tmp_path_factory.mktemp("oauth2_dcr_rate_limited") / "phoenix.db"),
        extra={"PHOENIX_OAUTH2_DCR_RATE_LIMIT_PER_HOUR": "1"},
    )
    with _server(_AppInfo(env)) as app:
        yield app


@pytest.fixture
def _app_dcr_enabled(
    _ports: Iterator[int],
    tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[_AppInfo]:
    port = next(_ports)
    env = _oauth2_app_env(
        port=port,
        grpc_port=next(_ports),
        database=str(tmp_path_factory.mktemp("oauth2_dcr_enabled") / "phoenix.db"),
        extra={
            "PHOENIX_DISABLE_RATE_LIMIT": "true",
            "PHOENIX_OAUTH2_DYNAMIC_CLIENT_REGISTRATION": "enabled",
            "PHOENIX_OAUTH2_ALLOWED_REDIRECT_HOSTS": "vscode.dev,insiders.vscode.dev",
        },
    )
    with _server(_AppInfo(env)) as app:
        yield app


_SHORT_GRANT_EXPIRY_SECONDS = 120


@pytest.fixture
def _app_short_grant(
    _ports: Iterator[int],
    tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[_AppInfo]:
    """An app whose grant ceiling is shorter than its default token lifetimes.

    Grants default to 90 days and access tokens to 10 minutes, so token lifetimes never
    reach the ceiling and clamping is unobservable. Squeezing the ceiling below both makes
    it the binding constraint.
    """
    port = next(_ports)
    env = _oauth2_app_env(
        port=port,
        grpc_port=next(_ports),
        database=str(tmp_path_factory.mktemp("oauth2_short_grant") / "phoenix.db"),
        extra={
            "PHOENIX_DISABLE_RATE_LIMIT": "true",
            "PHOENIX_OAUTH2_GRANT_EXPIRY_DAYS": str(_SHORT_GRANT_EXPIRY_SECONDS / 86400),
        },
    )
    with _server(_AppInfo(env)) as app:
        yield app


@pytest.fixture
def _app_dcr_disabled(
    _ports: Iterator[int],
    tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[_AppInfo]:
    port = next(_ports)
    env = _oauth2_app_env(
        port=port,
        grpc_port=next(_ports),
        database=str(tmp_path_factory.mktemp("oauth2_dcr_disabled") / "phoenix.db"),
        extra={
            "PHOENIX_DISABLE_RATE_LIMIT": "true",
            "PHOENIX_OAUTH2_DYNAMIC_CLIENT_REGISTRATION": "disabled",
            "PHOENIX_ENABLE_MCP_SERVER": "false",
        },
    )
    with _server(_AppInfo(env)) as app:
        yield app


def _oauth2_app_env(
    *,
    port: int,
    grpc_port: int,
    database: str,
    extra: Mapping[str, str],
) -> dict[str, str]:
    return {
        "PHOENIX_PORT": str(port),
        "PHOENIX_GRPC_PORT": str(grpc_port),
        "PHOENIX_MASK_INTERNAL_SERVER_ERRORS": "false",
        "PHOENIX_SQL_DATABASE_URL": f"sqlite:///{database}",
        "PHOENIX_ENABLE_AUTH": "true",
        "PHOENIX_SECRET": token_hex(16),
        "PHOENIX_ADMIN_SECRET": token_hex(16),
        "PHOENIX_CSRF_TRUSTED_ORIGINS": f",http://localhost,http://127.0.0.1:{port},",
        **extra,
    }


def _origin(url: str) -> str:
    parsed = urlparse(url)
    return f"{parsed.scheme}://{parsed.netloc}"
