litellm/tests/unit/proxy/credential_endpoints/test_endpoints.py

"""Tests for the credential management endpoints."""

import json
from contextlib import contextmanager
from typing import Final
from unittest.mock import AsyncMock, MagicMock, patch

import httpx
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import ec
from fastapi import HTTPException, Request
from fastapi.testclient import TestClient

import litellm
from litellm.models.credentials import CredentialSource
from litellm.proxy._types import LiteLLM_UserTable, LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.credential_endpoints.endpoints import get_llm_router
from litellm.proxy.proxy_server import app
from litellm.types.utils import CredentialItem

client = TestClient(app)


def _as_admin():
    return UserAPIKeyAuth(api_key="test-key", user_role="proxy_admin")


def _as_non_admin():
    return UserAPIKeyAuth(api_key="test-key", user_role="internal_user")


def _call_as(method: str, path: str, json_body: dict | None = None, auth=_as_admin):
    missing = object()
    previous_override = app.dependency_overrides.get(user_api_key_auth, missing)
    app.dependency_overrides[user_api_key_auth] = auth
    try:
        return client.request(method, path, json=json_body, headers={"Authorization": "Bearer test-key"})
    finally:
        if previous_override is missing:
            app.dependency_overrides.pop(user_api_key_auth, None)
        else:
            app.dependency_overrides[user_api_key_auth] = previous_override


def _patch_credential(name: str, body: dict, auth=_as_admin):
    return _call_as("PATCH", f"/credentials/{name}", body, auth)


def _post_credential(body: dict, auth=_as_admin):
    return _call_as("POST", "/credentials", body, auth)


def _delete_credential(name: str, auth=_as_admin):
    return _call_as("DELETE", f"/credentials/{name}", auth=auth)


def _list_credentials():
    return _call_as("GET", "/credentials")


def _prisma_without_credential_rows() -> MagicMock:
    prisma_client = MagicMock()
    prisma_client.db.litellm_credentialstable.find_unique = AsyncMock(return_value=None)
    return prisma_client


@pytest.fixture
def credential_store():
    """Stands the credential store up for one test: whether the database is reachable, what
    the proxy is already serving from memory, which router deployments resolve against, and
    what each repository call hands back."""

    def install(
        *,
        connected: bool = True,
        in_memory: tuple[object, ...] = (),
        llm_router: object | None = None,
        **repository_calls: AsyncMock,
    ) -> None:
        patch(
            "litellm.proxy.proxy_server.prisma_client", _prisma_without_credential_rows() if connected else None
        ).start()
        patch("litellm.proxy.proxy_server.master_key", "sk-test-master").start()
        patch.object(litellm, "credential_list", list(in_memory)).start()
        app.dependency_overrides[get_llm_router] = lambda: llm_router
        repository = patch("litellm.proxy.credential_endpoints.endpoints.CredentialsRepository").start()
        repository.return_value.find_by_name = AsyncMock(return_value=None)
        for call_name, result in repository_calls.items():
            setattr(repository.return_value, call_name, result)

    yield install
    patch.stopall()
    app.dependency_overrides.pop(get_llm_router, None)


@contextmanager
def _repository_holding(stored: CredentialItem | None):
    """The credentials repository seam, answering ``find_by_name`` with ``stored`` and recording
    the writes the handler attempts. Patched at both import sites, since the handlers resolve an
    existing credential through ``hydrate_named_credential`` (memory first, then this repository)
    and then write through their own ``CredentialsRepository`` binding."""
    with (
        patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.proxy_server.prisma_client", MagicMock()
        ),
        patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.proxy_server.master_key", "sk-test-master"
        ),
        patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.credential_endpoints.endpoints.CredentialsRepository"
        ) as repository,
        patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.common_utils.credential_hydration.CredentialsRepository", repository
        ),
    ):
        repository.return_value.find_by_name = AsyncMock(return_value=stored)
        repository.return_value.create = AsyncMock(return_value=None)
        repository.return_value.update_by_name = AsyncMock(return_value=None)
        repository.return_value.delete_by_name = AsyncMock(return_value=stored)
        yield repository.return_value


def test_create_credential_write_omits_the_patch_only_deletion_field(restore_credential_list):
    """Regression: CredentialItem.credential_values_to_delete is a PATCH-only field that
    defaults to None on every other construction path. A bare .model_dump() (without
    exclude_none) on the create path put a `credential_values_to_delete: null` key into the
    Prisma write, which litellm_credentialstable has no column for."""
    with _repository_holding(None) as repository:
        response = _post_credential(
            {
                "credential_name": "new-cred",
                "credential_values": {"api_key": "sk-new"},
                "credential_info": {"custom_llm_provider": "openai"},
            }
        )

    assert response.status_code == 200, response.text
    written_data = repository.create.await_args.kwargs["data"]
    assert "credential_values_to_delete" not in written_data


def test_update_credential_answers_404_when_the_credential_does_not_exist(credential_store):
    """Regression: the handler used to ``return handle_exception_on_proxy(e)``, which makes
    the exception the response body and lets FastAPI answer 200, so a write the handler
    rejected read as a success to every caller that checks the status. The dashboard's API
    client branches on the status, so it reported a failed edit as applied."""
    credential_store(find_by_name=AsyncMock(return_value=None))

    response = _patch_credential(
        "definitely-not-there",
        {"credential_name": "definitely-not-there", "credential_values": {"api_key": "sk-x"}, "credential_info": {}},
    )

    assert response.status_code == 404, f"rejected write answered {response.status_code}: {response.text}"
    assert "error" in response.json()


def test_update_credential_answers_500_when_the_database_is_not_connected(credential_store):
    """The other rejection this handler raises must carry its own status too."""
    credential_store(connected=False)

    response = _patch_credential(
        "any-name",
        {"credential_name": "any-name", "credential_values": {"api_key": "sk-x"}, "credential_info": {}},
    )

    assert response.status_code == 500, f"rejected write answered {response.status_code}: {response.text}"


def test_update_credential_still_answers_200_on_a_successful_write(credential_store):
    """The fix must not turn a legitimate update into an error; the dashboard and the
    Playwright credentials spec both assert the success path."""
    stored = CredentialItem(
        credential_name="existing",
        credential_values={"api_key": "sk-old"},
        credential_info={"custom_llm_provider": "openai"},
    )
    credential_store(find_by_name=AsyncMock(return_value=stored), update_by_name=AsyncMock(return_value=None))

    response = _patch_credential(
        "existing",
        {"credential_name": "existing", "credential_values": {"api_key": "sk-new"}, "credential_info": {}},
    )

    assert response.status_code == 200, response.text
    assert response.json()["success"] is True


def _get_jwks(name: str):
    return _call_as("GET", f"/credentials/{name}/jwks")


@pytest.fixture
def restore_credential_list(monkeypatch):
    monkeypatch.setattr(litellm, "credential_list", [])


def test_update_credential_rejects_overlap_between_update_and_delete():
    """A key in both sets is ambiguous (set to what value, before or after the delete?), so the
    endpoint must reject it outright rather than picking a resolution order silently."""
    response = _patch_credential(
        "any-name",
        {
            "credential_name": "any-name",
            "credential_values": {"api_key": "sk-new"},
            "credential_values_to_delete": ["api_key"],
            "credential_info": {},
        },
    )

    assert response.status_code == 400, response.text
    assert "api_key" in response.json()["error"]["message"]


def test_update_credential_deletion_removes_the_key_from_the_db_write(restore_credential_list):
    """The bug this closes: switching WIF identity sources (or WIF -> api_key) left the old
    variant's fields behind in the DB row, which wif.py then rejects by presence."""
    stored = CredentialItem(
        credential_name="wif-cred",
        credential_values={"anthropic_identity_source": "keycloak", "anthropic_keycloak_client_id": "old-client"},
        credential_info={"custom_llm_provider": "anthropic"},
    )
    with (
        patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.proxy_server.prisma_client", MagicMock()
        ),
        patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.credential_endpoints.endpoints.CredentialsRepository"
        ) as repository,  # test-quality-ok: the proxy wiring under test is what this patches
    ):
        repository.return_value.find_by_name = AsyncMock(return_value=stored)
        update_mock = AsyncMock(return_value=None)
        repository.return_value.update_by_name = update_mock

        response = _patch_credential(
            "wif-cred",
            {
                "credential_name": "wif-cred",
                "credential_values": {},
                "credential_values_to_delete": ["anthropic_keycloak_client_id"],
                "credential_info": {},
            },
        )

    assert response.status_code == 200, response.text
    written_values = json.loads(update_mock.await_args.kwargs["data"]["credential_values"])
    assert "anthropic_keycloak_client_id" not in written_values
    assert written_values["anthropic_identity_source"] == "keycloak"


def test_update_credential_deletion_updates_in_memory_credential_list(restore_credential_list, monkeypatch):
    """The in-memory list is what the request-time auth resolvers read; a deletion that only
    landed in the DB would leave the stale field servable until the next process restart."""
    monkeypatch.setattr(
        litellm,
        "credential_list",
        [
            CredentialItem(
                credential_name="wif-cred",
                credential_values={
                    "anthropic_identity_source": "keycloak",
                    "anthropic_keycloak_client_id": "old-client",
                },
                credential_info={"custom_llm_provider": "anthropic"},
            )
        ],
    )
    stored = CredentialItem(
        credential_name="wif-cred",
        credential_values={"anthropic_identity_source": "keycloak", "anthropic_keycloak_client_id": "old-client"},
        credential_info={"custom_llm_provider": "anthropic"},
    )
    with (
        patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.proxy_server.prisma_client", MagicMock()
        ),
        patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.credential_endpoints.endpoints.CredentialsRepository"
        ) as repository,  # test-quality-ok: the proxy wiring under test is what this patches
    ):
        repository.return_value.find_by_name = AsyncMock(return_value=stored)
        repository.return_value.update_by_name = AsyncMock(return_value=None)

        response = _patch_credential(
            "wif-cred",
            {
                "credential_name": "wif-cred",
                "credential_values": {},
                "credential_values_to_delete": ["anthropic_keycloak_client_id"],
                "credential_info": {},
            },
        )

    assert response.status_code == 200, response.text
    in_memory = next(c for c in litellm.credential_list if c.credential_name == "wif-cred")
    assert "anthropic_keycloak_client_id" not in in_memory.credential_values
    assert in_memory.credential_values["anthropic_identity_source"] == "keycloak"


def test_update_credential_leaves_untouched_fields_alone():
    """Regression for the masked-value hazard: GET /credentials masks values, so a PATCH that
    only names the field being changed must not let an untouched field be nulled or overwritten
    by anything a round-tripped (masked) form value could contain."""
    stored = CredentialItem(
        credential_name="existing",
        credential_values={"api_key": "sk-real-value", "api_base": "https://api.anthropic.com"},
        credential_info={"custom_llm_provider": "anthropic"},
    )
    with (
        patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.proxy_server.prisma_client", MagicMock()
        ),
        patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.proxy_server.master_key", "sk-test-master"
        ),
        patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.credential_endpoints.endpoints.CredentialsRepository"
        ) as repository,  # test-quality-ok: the proxy wiring under test is what this patches
    ):
        repository.return_value.find_by_name = AsyncMock(return_value=stored)
        update_mock = AsyncMock(return_value=None)
        repository.return_value.update_by_name = update_mock

        response = _patch_credential(
            "existing",
            {"credential_name": "existing", "credential_values": {"api_key": "sk-rotated"}, "credential_info": {}},
        )

    assert response.status_code == 200, response.text
    written_values = json.loads(update_mock.await_args.kwargs["data"]["credential_values"])
    assert written_values["api_base"] == "https://api.anthropic.com"


def test_create_credential_never_stores_a_null_credential_value(restore_credential_list):
    """The dashboard posts a key for every field on the provider's form, and the ones the operator
    left blank arrive as null. A null carries no credential, and the federation resolver refuses a
    foreign variant's field by key, so a stored null wedges every deployment naming this credential."""
    with _repository_holding(None) as repository:
        response = _post_credential(
            {
                "credential_name": "new-cred",
                "credential_values": {"api_key": "sk-new", "anthropic_issuer_url": None},
                "credential_info": {"custom_llm_provider": "anthropic"},
            }
        )

    assert response.status_code == 200, response.text
    written_values = json.loads(repository.create.await_args.kwargs["data"]["credential_values"])
    assert "anthropic_issuer_url" not in written_values
    assert "api_key" in written_values


def test_update_credential_never_stores_a_null_credential_value(restore_credential_list):
    """Same null on the update path, where the merge writes the whole row back: the field the null
    named keeps whatever it stored, since removing a field is what credential_values_to_delete is for."""
    stored = CredentialItem(
        credential_name="wif-cred",
        credential_values={"anthropic_identity_source": "keycloak", "anthropic_keycloak_client_id": "old-client"},
        credential_info={"custom_llm_provider": "anthropic"},
    )
    with _repository_holding(stored) as repository:
        response = _patch_credential(
            "wif-cred",
            {
                "credential_name": "wif-cred",
                "credential_values": {"anthropic_keycloak_client_id": None},
                "credential_info": {},
            },
        )

    assert response.status_code == 200, response.text
    written_values = json.loads(repository.update_by_name.await_args.kwargs["data"]["credential_values"])
    assert written_values["anthropic_keycloak_client_id"] == "old-client"


def test_update_credential_never_syncs_a_null_into_the_in_memory_credential(restore_credential_list, monkeypatch):
    """The in-memory list is what request-time resolution reads, so a null that only got kept out of
    the DB row would still wedge every deployment until the next restart."""
    in_memory = CredentialItem(
        credential_name="plain-cred",
        credential_values={"api_key": "sk-old"},
        credential_info={"custom_llm_provider": "anthropic"},
    )
    monkeypatch.setattr(litellm, "credential_list", [in_memory])
    with _repository_holding(
        CredentialItem(
            credential_name="plain-cred",
            credential_values={"api_key": "sk-old"},
            credential_info={"custom_llm_provider": "anthropic"},
        )
    ):
        response = _patch_credential(
            "plain-cred",
            {
                "credential_name": "plain-cred",
                "credential_values": {"api_key": "sk-rotated", "anthropic_issuer_url": None},
                "credential_info": {},
            },
        )

    assert response.status_code == 200, response.text
    synced = next(c for c in litellm.credential_list if c.credential_name == "plain-cred")
    assert "anthropic_issuer_url" not in synced.credential_values
    assert synced.credential_values["api_key"] == "sk-rotated"


def _generate_es256_pem() -> str:
    key = ec.generate_private_key(ec.SECP256R1())
    return key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    ).decode()


class TestCredentialJwksExport:
    def test_jwks_export_succeeds_for_an_internal_issuer_credential(self, restore_credential_list, monkeypatch):
        monkeypatch.setenv("JWKS_TEST_SIGNING_KEY", _generate_es256_pem())
        monkeypatch.setattr(
            litellm,
            "credential_list",
            [
                CredentialItem(
                    credential_name="anthropic-issuer",
                    credential_values={
                        "anthropic_identity_source": "internal_issuer",
                        "anthropic_issuer_url": "https://issuer.example.com",
                        "anthropic_issuer_subject": "my-workload",
                        "anthropic_issuer_signing_key_ref": "os.environ/JWKS_TEST_SIGNING_KEY",
                    },
                    credential_info={"custom_llm_provider": "anthropic"},
                )
            ],
        )

        response = _get_jwks("anthropic-issuer")

        assert response.status_code == 200, response.text
        body = response.json()
        assert body["keys"][0]["kty"] == "EC"
        assert body["keys"][0]["crv"] == "P-256"
        # The private key material must never leave the process via this endpoint.
        assert "JWKS_TEST_SIGNING_KEY" not in response.text
        assert "PRIVATE KEY" not in response.text

    def test_jwks_export_treats_blank_optional_fields_as_unset(self, restore_credential_list, monkeypatch):
        monkeypatch.setenv("JWKS_TEST_SIGNING_KEY", _generate_es256_pem())
        monkeypatch.setattr(
            litellm,
            "credential_list",
            [
                CredentialItem(
                    credential_name="anthropic-issuer-blanks",
                    credential_values={
                        "anthropic_identity_source": "internal_issuer",
                        "anthropic_issuer_url": "https://issuer.example.com",
                        "anthropic_issuer_subject": "my-workload",
                        "anthropic_issuer_signing_key_ref": "os.environ/JWKS_TEST_SIGNING_KEY",
                        "anthropic_issuer_audience": "",
                        "anthropic_issuer_ttl_seconds": "",
                    },
                    credential_info={"custom_llm_provider": "anthropic"},
                )
            ],
        )

        response = _get_jwks("anthropic-issuer-blanks")

        assert response.status_code == 200, response.text
        assert response.json()["keys"][0]["kty"] == "EC"

    def test_jwks_export_accepts_the_dashboard_provider_casing(self, restore_credential_list, monkeypatch):
        monkeypatch.setenv("JWKS_TEST_SIGNING_KEY", _generate_es256_pem())
        monkeypatch.setattr(
            litellm,
            "credential_list",
            [
                CredentialItem(
                    credential_name="anthropic-from-modal",
                    credential_values={
                        "anthropic_identity_source": "internal_issuer",
                        "anthropic_issuer_url": "https://issuer.example.com",
                        "anthropic_issuer_subject": "my-workload",
                        "anthropic_issuer_signing_key_ref": "os.environ/JWKS_TEST_SIGNING_KEY",
                    },
                    credential_info={"custom_llm_provider": "Anthropic"},
                )
            ],
        )

        response = _get_jwks("anthropic-from-modal")

        assert response.status_code == 200, response.text
        assert response.json()["keys"][0]["kty"] == "EC"

    def test_jwks_export_404s_for_a_non_anthropic_credential(self, restore_credential_list, monkeypatch):
        monkeypatch.setattr(
            litellm,
            "credential_list",
            [
                CredentialItem(
                    credential_name="openai-key",
                    credential_values={"api_key": "sk-x"},
                    credential_info={"custom_llm_provider": "openai"},
                )
            ],
        )

        response = _get_jwks("openai-key")

        assert response.status_code == 404, response.text

    def test_jwks_export_404s_for_an_anthropic_credential_without_internal_issuer(
        self, restore_credential_list, monkeypatch
    ):
        monkeypatch.setattr(
            litellm,
            "credential_list",
            [
                CredentialItem(
                    credential_name="anthropic-apikey",
                    credential_values={"api_key": "sk-ant"},
                    credential_info={"custom_llm_provider": "anthropic"},
                )
            ],
        )

        response = _get_jwks("anthropic-apikey")

        assert response.status_code == 404, response.text

    def test_jwks_export_404s_for_an_unknown_credential(self, restore_credential_list):
        with patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.proxy_server.prisma_client", None
        ):  # test-quality-ok: the proxy wiring under test is what this patches
            response = _get_jwks("does-not-exist")

        assert response.status_code == 404, response.text

    def test_jwks_export_requires_proxy_admin(self, restore_credential_list, monkeypatch):
        monkeypatch.setenv("JWKS_TEST_SIGNING_KEY", _generate_es256_pem())
        monkeypatch.setattr(
            litellm,
            "credential_list",
            [
                CredentialItem(
                    credential_name="anthropic-issuer",
                    credential_values={
                        "anthropic_identity_source": "internal_issuer",
                        "anthropic_issuer_url": "https://issuer.example.com",
                        "anthropic_issuer_subject": "my-workload",
                        "anthropic_issuer_signing_key_ref": "os.environ/JWKS_TEST_SIGNING_KEY",
                    },
                    credential_info={"custom_llm_provider": "anthropic"},
                )
            ],
        )

        def _as_internal_user():
            return UserAPIKeyAuth(api_key="test-key", user_role="internal_user")

        app.dependency_overrides[user_api_key_auth] = _as_internal_user
        try:
            response = client.get("/credentials/anthropic-issuer/jwks", headers={"Authorization": "Bearer test-key"})
        finally:
            app.dependency_overrides.pop(user_api_key_auth, None)

        assert response.status_code == 403, response.text


class TestNonAdminCannotPersistWifFieldsOnCredential:
    """A credential's ``credential_values`` feeds the same WIF resolution as a deployment's own
    ``litellm_params`` when referenced by ``litellm_credential_name``. A non-admin must not be
    able to create or update a credential carrying a server-owned WIF field such as
    ``anthropic_keycloak_token_url`` (destination) or ``anthropic_keycloak_client_secret_ref``
    (which secret to read and send there)."""

    def test_non_admin_cannot_create_a_credential_with_a_wif_destination(self):
        with patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.proxy_server.prisma_client", MagicMock()
        ):  # test-quality-ok: the proxy wiring under test is what this patches
            response = _post_credential(
                {
                    "credential_name": "attacker-cred",
                    "credential_values": {"anthropic_keycloak_token_url": "https://evil.example.com/token"},
                    "credential_info": {"custom_llm_provider": "anthropic"},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 403, response.text
        assert "anthropic_keycloak_token_url" in response.json()["error"]["message"]

    def test_non_admin_cannot_create_a_credential_with_a_wif_secret_ref(self):
        with patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.proxy_server.prisma_client", MagicMock()
        ):  # test-quality-ok: the proxy wiring under test is what this patches
            response = _post_credential(
                {
                    "credential_name": "attacker-cred",
                    "credential_values": {"anthropic_keycloak_client_secret_ref": "os.environ/LITELLM_MASTER_KEY"},
                    "credential_info": {"custom_llm_provider": "anthropic"},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 403, response.text

    def test_non_admin_can_create_a_credential_without_wif_fields(self, restore_credential_list):
        with _repository_holding(None) as repository:
            response = _post_credential(
                {
                    "credential_name": "ordinary-cred",
                    "credential_values": {"api_key": "sk-new"},
                    "credential_info": {"custom_llm_provider": "openai"},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 200, response.text
        repository.create.assert_awaited_once()

    def test_proxy_admin_can_create_a_credential_with_a_wif_destination(self, restore_credential_list):
        with _repository_holding(None) as repository:
            response = _post_credential(
                {
                    "credential_name": "admin-cred",
                    "credential_values": {"anthropic_keycloak_token_url": "https://keycloak.internal/token"},
                    "credential_info": {"custom_llm_provider": "anthropic"},
                },
                auth=_as_admin,
            )

        assert response.status_code == 200, response.text
        repository.create.assert_awaited_once()

    def test_non_admin_cannot_create_a_credential_with_oauth_token_exchange_endpoint(self, restore_credential_list):
        with _repository_holding(None) as repository:
            response = _post_credential(
                {
                    "credential_name": "attacker-oauth",
                    "credential_values": {"token_exchange_endpoint": "https://attacker.example/token"},
                    "credential_info": {"custom_llm_provider": "microsoft_365_copilot"},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 403, response.text
        assert "token_exchange_endpoint" in response.json()["error"]["message"]
        repository.create.assert_not_awaited()

    def test_proxy_admin_can_create_a_credential_with_oauth_token_exchange_endpoint(self, restore_credential_list):
        with _repository_holding(None) as repository:
            response = _post_credential(
                {
                    "credential_name": "admin-oauth",
                    "credential_values": {"token_exchange_endpoint": "https://identity.example.com/token"},
                    "credential_info": {"custom_llm_provider": "microsoft_365_copilot"},
                },
                auth=_as_admin,
            )

        assert response.status_code == 200, response.text
        repository.create.assert_awaited_once()

    def test_non_admin_cannot_create_a_credential_with_an_openai_token_file(self):
        with patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.proxy_server.prisma_client", MagicMock()
        ):
            response = _post_credential(
                {
                    "credential_name": "attacker-cred",
                    "credential_values": {"openai_identity_token_file": "/var/run/secrets/tokens/attacker"},
                    "credential_info": {"custom_llm_provider": "openai"},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 403, response.text
        assert "openai_identity_token_file" in response.json()["error"]["message"]

    def test_proxy_admin_can_create_a_credential_with_the_openai_identity_trio(self, restore_credential_list):
        with _repository_holding(None) as repository:
            response = _post_credential(
                {
                    "credential_name": "openai-wif",
                    "credential_values": {
                        "openai_identity_provider_id": "idp_1",
                        "openai_service_account_id": "user-1",
                        "openai_identity_token_file": "/var/run/secrets/tokens/openai",
                    },
                    "credential_info": {"custom_llm_provider": "openai"},
                },
                auth=_as_admin,
            )

        assert response.status_code == 200, response.text
        repository.create.assert_awaited_once()

    def test_non_admin_cannot_update_a_credential_to_add_a_wif_destination(self):
        stored = CredentialItem(
            credential_name="existing",
            credential_values={"api_key": "sk-old"},
            credential_info={"custom_llm_provider": "anthropic"},
        )
        with (
            patch(  # test-quality-ok: the proxy wiring under test is what this patches
                "litellm.proxy.proxy_server.prisma_client", MagicMock()
            ),
            patch(  # test-quality-ok: the proxy wiring under test is what this patches
                "litellm.proxy.credential_endpoints.endpoints.CredentialsRepository"
            ) as repository,  # test-quality-ok: the proxy wiring under test is what this patches
        ):
            repository.return_value.find_by_name = AsyncMock(return_value=stored)
            update_mock = AsyncMock(return_value=None)
            repository.return_value.update_by_name = update_mock

            response = _patch_credential(
                "existing",
                {
                    "credential_name": "existing",
                    "credential_values": {"anthropic_keycloak_token_url": "https://evil.example.com/token"},
                    "credential_info": {},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 403, response.text
        update_mock.assert_not_awaited()

    def test_non_admin_cannot_patch_wif_fields_onto_a_credential_through_model_id(self, credential_store):
        """Regression: the PATCH gate read only the submitted ``credential_values``, so a non-admin
        naming a federated deployment through ``model_id`` had its WIF fields copied onto an
        ordinary credential unchecked, while POST already gated the resolved values."""
        stored = CredentialItem(credential_name="existing", credential_values={"api_key": "sk-old"}, credential_info={})
        update_by_name = AsyncMock(return_value=None)
        router = MagicMock()
        router.get_deployment.return_value = {"model_name": "claude-opus-5-5"}
        router.get_deployment_credentials.return_value = {
            "anthropic_keycloak_token_url": "https://keycloak.internal/token",
            "anthropic_keycloak_client_secret_ref": "os.environ/KEYCLOAK_CLIENT_SECRET",
        }
        credential_store(find_by_name=AsyncMock(return_value=stored), update_by_name=update_by_name, llm_router=router)

        response = _patch_credential(
            "existing",
            {"credential_name": "existing", "model_id": "federated-deployment", "credential_info": {}},
            auth=_as_non_admin,
        )

        assert response.status_code == 403, response.text
        update_by_name.assert_not_awaited()

    def test_proxy_admin_can_patch_wif_fields_onto_a_credential_through_model_id(self, credential_store):
        stored = CredentialItem(credential_name="existing", credential_values={"api_key": "sk-old"}, credential_info={})
        update_by_name = AsyncMock(return_value=None)
        router = MagicMock()
        router.get_deployment.return_value = {"model_name": "claude-opus-5-5"}
        router.get_deployment_credentials.return_value = {
            "anthropic_keycloak_token_url": "https://keycloak.internal/token"
        }
        credential_store(find_by_name=AsyncMock(return_value=stored), update_by_name=update_by_name, llm_router=router)

        response = _patch_credential(
            "existing",
            {"credential_name": "existing", "model_id": "federated-deployment", "credential_info": {}},
            auth=_as_admin,
        )

        assert response.status_code == 200, response.text
        written = json.loads(update_by_name.await_args.kwargs["data"]["credential_values"])
        assert "anthropic_keycloak_token_url" in written, "the deployment's WIF field reaches the stored credential"

    def test_proxy_admin_can_update_a_credential_to_add_a_wif_destination(self):
        stored = CredentialItem(
            credential_name="existing",
            credential_values={"api_key": "sk-old"},
            credential_info={"custom_llm_provider": "anthropic"},
        )
        with (
            patch(  # test-quality-ok: the proxy wiring under test is what this patches
                "litellm.proxy.proxy_server.prisma_client", MagicMock()
            ),
            patch(  # test-quality-ok: the proxy wiring under test is what this patches
                "litellm.proxy.proxy_server.master_key", "sk-test-master"
            ),
            patch(  # test-quality-ok: the proxy wiring under test is what this patches
                "litellm.proxy.credential_endpoints.endpoints.CredentialsRepository"
            ) as repository,  # test-quality-ok: the proxy wiring under test is what this patches
        ):
            repository.return_value.find_by_name = AsyncMock(return_value=stored)
            update_mock = AsyncMock(return_value=None)
            repository.return_value.update_by_name = update_mock

            response = _patch_credential(
                "existing",
                {
                    "credential_name": "existing",
                    "credential_values": {"anthropic_keycloak_token_url": "https://keycloak.internal/token"},
                    "credential_info": {},
                },
                auth=_as_admin,
            )

        assert response.status_code == 200, response.text
        update_mock.assert_awaited_once()


def _wif_credential(name: str = "federated-cred", source: CredentialSource = "db") -> CredentialItem:
    return CredentialItem(
        credential_name=name,
        credential_values={
            "anthropic_keycloak_token_url": "https://keycloak.internal/token",
            "api_key": "sk-old",
        },
        credential_info={"custom_llm_provider": "anthropic"},
        source=source,
    )


def _plain_credential(name: str = "ordinary-cred") -> CredentialItem:
    return CredentialItem(
        credential_name=name,
        credential_values={"api_key": "sk-old"},
        credential_info={"custom_llm_provider": "openai"},
    )


class TestNonAdminCannotTouchAStoredWifCredential:
    """The WIF gate used to read only the incoming ``credential_values``, so a non-admin could
    drop a federation field by naming it in ``credential_values_to_delete`` (breaking every
    deployment that references the credential), or edit a stored admin-owned WIF credential by
    sending a payload carrying no WIF field at all. The gate is evaluated against the effective
    surface of the operation: incoming keys (a ``null`` value still persists the key), deleted
    keys, and the stored credential, wherever it lives (DB row or config-only ``credential_list``
    entry)."""

    def test_non_admin_cannot_delete_a_wif_field_off_a_credential(self, restore_credential_list):
        with _repository_holding(_plain_credential("some-cred")) as repository:
            response = _patch_credential(
                "some-cred",
                {
                    "credential_name": "some-cred",
                    "credential_values": {},
                    "credential_values_to_delete": ["anthropic_keycloak_token_url"],
                    "credential_info": {},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 403, response.text
        assert "anthropic_keycloak_token_url" in response.text
        repository.update_by_name.assert_not_awaited()

    def test_non_admin_cannot_patch_a_stored_wif_credential(self, restore_credential_list):
        with _repository_holding(_wif_credential("federated-cred")) as repository:
            response = _patch_credential(
                "federated-cred",
                {
                    "credential_name": "federated-cred",
                    "credential_values": {"api_key": "sk-attacker"},
                    "credential_info": {},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 403, response.text
        assert "anthropic_keycloak_token_url" in response.text
        repository.update_by_name.assert_not_awaited()

    def test_proxy_admin_can_delete_a_wif_field_off_a_credential(self, restore_credential_list):
        with _repository_holding(_wif_credential("federated-cred")) as repository:
            response = _patch_credential(
                "federated-cred",
                {
                    "credential_name": "federated-cred",
                    "credential_values": {},
                    "credential_values_to_delete": ["anthropic_keycloak_token_url"],
                    "credential_info": {},
                },
                auth=_as_admin,
            )

        assert response.status_code == 200, response.text
        written_values = json.loads(repository.update_by_name.await_args.kwargs["data"]["credential_values"])
        assert "anthropic_keycloak_token_url" not in written_values

    def test_proxy_admin_can_patch_a_stored_wif_credential(self, restore_credential_list):
        with _repository_holding(_wif_credential("federated-cred")) as repository:
            response = _patch_credential(
                "federated-cred",
                {
                    "credential_name": "federated-cred",
                    "credential_values": {"api_key": "sk-rotated"},
                    "credential_info": {},
                },
                auth=_as_admin,
            )

        assert response.status_code == 200, response.text
        written_values = json.loads(repository.update_by_name.await_args.kwargs["data"]["credential_values"])
        assert written_values["anthropic_keycloak_token_url"] is not None

    def test_non_admin_can_still_patch_a_credential_with_no_wif_fields_anywhere(self, restore_credential_list):
        with _repository_holding(_plain_credential("ordinary-cred")) as repository:
            response = _patch_credential(
                "ordinary-cred",
                {
                    "credential_name": "ordinary-cred",
                    "credential_values": {"api_key": "sk-rotated"},
                    "credential_info": {},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 200, response.text
        repository.update_by_name.assert_awaited_once()

    def test_non_admin_cannot_delete_a_stored_wif_credential(self, restore_credential_list):
        """DELETE takes the whole row, so it drops the admin-owned federation settings as surely
        as a targeted key deletion would."""
        with _repository_holding(_wif_credential("federated-cred")) as repository:
            response = _delete_credential("federated-cred", auth=_as_non_admin)

        assert response.status_code == 403, response.text
        assert response.json()["error"]["param"] == "anthropic_keycloak_token_url"
        repository.delete_by_name.assert_not_awaited()

    def test_a_stale_in_memory_copy_does_not_authorize_deleting_a_stored_wif_credential(
        self, restore_credential_list, monkeypatch
    ):
        """Resolution reads memory first and stops, which is right when serving a request. A pod
        whose in-memory copy predates an admin adding the federation fields must not read that
        stale object and authorize the delete: the gate takes the union of memory and the row."""
        monkeypatch.setattr(litellm, "credential_list", [_plain_credential("federated-cred")])

        with _repository_holding(_wif_credential("federated-cred")) as repository:
            response = _delete_credential("federated-cred", auth=_as_non_admin)

        assert response.status_code == 403, response.text
        assert response.json()["error"]["param"] == "anthropic_keycloak_token_url"
        repository.delete_by_name.assert_not_awaited()

    def test_proxy_admin_can_delete_a_stored_wif_credential(self, restore_credential_list):
        with _repository_holding(_wif_credential("federated-cred")) as repository:
            response = _delete_credential("federated-cred", auth=_as_admin)

        assert response.status_code == 200, response.text
        repository.delete_by_name.assert_awaited_once_with("federated-cred")

    def test_non_admin_can_still_delete_a_credential_with_no_wif_fields(self, restore_credential_list):
        with _repository_holding(_plain_credential("ordinary-cred")) as repository:
            response = _delete_credential("ordinary-cred", auth=_as_non_admin)

        assert response.status_code == 200, response.text
        repository.delete_by_name.assert_awaited_once_with("ordinary-cred")

    def test_non_admin_cannot_null_out_a_wif_field_on_a_credential(self, restore_credential_list):
        """A JSON ``null`` still lands as a key in ``credential_values``. ``get_litellm_params``
        forwards a WIF kwarg on key presence and the federation resolver rejects a foreign
        variant's field by key, so a value-based gate let a non-admin persist the key and wedge
        every deployment referencing the credential at request time."""
        with _repository_holding(_plain_credential("some-cred")) as repository:
            response = _patch_credential(
                "some-cred",
                {
                    "credential_name": "some-cred",
                    "credential_values": {"anthropic_issuer_url": None},
                    "credential_info": {},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 403, response.text
        assert "anthropic_issuer_url" in response.text
        repository.update_by_name.assert_not_awaited()

    def test_non_admin_cannot_patch_a_credential_storing_a_null_wif_field(self, restore_credential_list):
        stored = CredentialItem(
            credential_name="nulled-cred",
            credential_values={"anthropic_issuer_url": None, "api_key": "sk-old"},
            credential_info={"custom_llm_provider": "anthropic"},
        )
        with _repository_holding(stored) as repository:
            response = _patch_credential(
                "nulled-cred",
                {
                    "credential_name": "nulled-cred",
                    "credential_values": {"api_key": "sk-attacker"},
                    "credential_info": {},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 403, response.text
        assert "anthropic_issuer_url" in response.text
        repository.update_by_name.assert_not_awaited()

    def test_proxy_admin_can_null_out_a_wif_field_on_a_credential(self, restore_credential_list):
        with _repository_holding(_wif_credential("federated-cred")) as repository:
            response = _patch_credential(
                "federated-cred",
                {
                    "credential_name": "federated-cred",
                    "credential_values": {"anthropic_keycloak_token_url": None},
                    "credential_info": {},
                },
                auth=_as_admin,
            )

        assert response.status_code == 200, response.text
        repository.update_by_name.assert_awaited_once()

    def test_non_admin_cannot_delete_a_config_only_wif_credential(self, restore_credential_list, monkeypatch):
        """A ``credential_list`` entry from config.yaml has no DB row, so a gate that consulted
        only the DB let a non-admin evict the admin-owned federation settings from memory."""
        config_credential = _wif_credential("config-wif", source="config")
        monkeypatch.setattr(litellm, "credential_list", [config_credential])
        with _repository_holding(None) as repository:
            response = _delete_credential("config-wif", auth=_as_non_admin)

        assert response.status_code == 403, response.text
        assert response.json()["error"]["param"] == "anthropic_keycloak_token_url"
        repository.delete_by_name.assert_not_awaited()
        assert litellm.credential_list == [config_credential]

    def test_proxy_admin_can_delete_a_config_only_wif_credential(self, restore_credential_list, monkeypatch):
        """The gate lets the admin through to the row delete. The 400 that follows is the rule for
        every config-only credential (no row to delete, the entry is back on the next boot), so the
        in-memory entry stays put too."""
        config_credential = _wif_credential("config-wif", source="config")
        monkeypatch.setattr(litellm, "credential_list", [config_credential])
        with _repository_holding(None) as repository:
            response = _delete_credential("config-wif", auth=_as_admin)

        assert response.status_code == 405, response.text
        assert response.headers["allow"] == "GET"
        assert "defined in config" in response.json()["error"]["message"]
        repository.delete_by_name.assert_awaited_once_with("config-wif")
        assert litellm.credential_list == [config_credential]

    def test_non_admin_cannot_shadow_a_config_only_wif_credential(self, restore_credential_list, monkeypatch):
        """POST with the same name carries no WIF field and collides with no DB row, yet
        ``CredentialAccessor.upsert_credentials`` would replace the admin entry in memory and
        the periodic config sync would then make the takeover permanent."""
        config_credential = _wif_credential("config-wif", source="config")
        monkeypatch.setattr(litellm, "credential_list", [config_credential])
        with _repository_holding(None) as repository:
            response = _post_credential(
                {
                    "credential_name": "config-wif",
                    "credential_values": {"api_key": "sk-attacker"},
                    "credential_info": {"custom_llm_provider": "anthropic"},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 403, response.text
        assert "anthropic_keycloak_token_url" in response.text
        repository.create.assert_not_awaited()
        assert litellm.credential_list == [config_credential]
        assert litellm.credential_list[0].credential_values["api_key"] == "sk-old"

    def test_proxy_admin_can_post_over_a_config_only_wif_credential(self, restore_credential_list, monkeypatch):
        monkeypatch.setattr(litellm, "credential_list", [_wif_credential("config-wif", source="config")])
        with _repository_holding(None) as repository:
            response = _post_credential(
                {
                    "credential_name": "config-wif",
                    "credential_values": {"api_key": "sk-rotated"},
                    "credential_info": {"custom_llm_provider": "anthropic"},
                },
                auth=_as_admin,
            )

        assert response.status_code == 200, response.text
        repository.create.assert_awaited_once()
        assert litellm.credential_list[0].credential_values == {"api_key": "sk-rotated"}

    def test_non_admin_cannot_post_a_null_wif_field(self, restore_credential_list):
        """Same key-presence rule on the create path: ``{"anthropic_issuer_url": null}`` persists
        the key, and the resolver reacts to the key."""
        with _repository_holding(None) as repository:
            response = _post_credential(
                {
                    "credential_name": "nulled-cred",
                    "credential_values": {"anthropic_issuer_url": None, "api_key": "sk-new"},
                    "credential_info": {"custom_llm_provider": "anthropic"},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 403, response.text
        assert "anthropic_issuer_url" in response.text
        repository.create.assert_not_awaited()
        assert litellm.credential_list == []

    def test_non_admin_cannot_shadow_a_db_stored_wif_credential(self, restore_credential_list):
        """Same hole for a WIF credential another pod wrote to the DB before this pod's in-memory
        list caught up: the existing-credential lookup falls through to the DB."""
        with _repository_holding(_wif_credential("federated-cred")) as repository:
            response = _post_credential(
                {
                    "credential_name": "federated-cred",
                    "credential_values": {"api_key": "sk-attacker"},
                    "credential_info": {"custom_llm_provider": "anthropic"},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 403, response.text
        repository.create.assert_not_awaited()

    def test_non_admin_can_still_post_a_credential_with_no_wif_fields_anywhere(self, restore_credential_list):
        with _repository_holding(None) as repository:
            response = _post_credential(
                {
                    "credential_name": "ordinary-cred",
                    "credential_values": {"api_key": "sk-new"},
                    "credential_info": {"custom_llm_provider": "openai"},
                },
                auth=_as_non_admin,
            )

        assert response.status_code == 200, response.text
        repository.create.assert_awaited_once()
        assert litellm.credential_list[0].credential_name == "ordinary-cred"


class TestManagementReadsTheStoredCredential:
    """Serving a request reads memory first, which is right. A management operation cannot: on a
    pod whose in-memory copy predates another pod's update it would act on superseded values."""

    @pytest.mark.asyncio
    async def test_authoritative_hydrate_prefers_the_row_over_a_stale_memory_copy(self):
        import litellm
        from litellm.proxy.common_utils.credential_hydration import (
            hydrate_named_credential,
            hydrate_named_credential_authoritative,
        )
        from litellm.types.utils import CredentialItem

        stale = CredentialItem(
            credential_name="anthropic-wif",
            credential_values={"anthropic_issuer_url": "https://old.example.com"},
            credential_info={"custom_llm_provider": "anthropic"},
        )
        row = {
            "credential_name": "anthropic-wif",
            "credential_values": {"anthropic_issuer_url": "https://new.example.com"},
            "credential_info": {"custom_llm_provider": "anthropic"},
        }

        prisma = MagicMock()
        prisma.db.litellm_credentialstable.find_unique = AsyncMock(return_value=row)

        with patch.object(litellm, "credential_list", [stale]):  # test-quality-ok: the stale copy under test
            served = await hydrate_named_credential("anthropic-wif", prisma)
            managed = await hydrate_named_credential_authoritative("anthropic-wif", prisma)

        assert served is not None and served.credential_values["anthropic_issuer_url"] == "https://old.example.com"
        assert managed is not None and managed.credential_values["anthropic_issuer_url"] == "https://new.example.com"

    @pytest.mark.asyncio
    async def test_authoritative_hydrate_falls_back_to_memory_when_the_row_is_absent(self):
        import litellm
        from litellm.proxy.common_utils.credential_hydration import hydrate_named_credential_authoritative
        from litellm.types.utils import CredentialItem

        only_in_memory = CredentialItem(
            credential_name="config-yaml-credential",
            credential_values={"anthropic_issuer_url": "https://configured.example.com"},
            credential_info={"custom_llm_provider": "anthropic"},
        )
        prisma = MagicMock()
        prisma.db.litellm_credentialstable.find_unique = AsyncMock(return_value=None)

        with patch.object(
            litellm, "credential_list", [only_in_memory]
        ):  # test-quality-ok: the config.yaml fallback under test
            resolved = await hydrate_named_credential_authoritative("config-yaml-credential", prisma)

        assert resolved is not None
        assert resolved.credential_values["anthropic_issuer_url"] == "https://configured.example.com"


def test_delete_credential_answers_404_when_the_credential_does_not_exist(credential_store):
    """Regression: prisma's ``delete`` hands back None when the ``where`` clause matched no row
    instead of raising, and the handler never looked. Deleting a name that was never stored
    answered 200 "Credential deleted successfully", so an operator scripting cleanup could not
    tell a real deletion from a typo."""
    credential_store(delete_by_name=AsyncMock(return_value=None))

    response = _delete_credential("definitely-not-there")

    assert response.status_code == 404, (
        f"delete of a missing credential answered {response.status_code}: {response.text}"
    )
    assert "definitely-not-there" in response.json()["error"]["message"]


def test_delete_credential_still_answers_200_and_drops_the_credential_from_memory(credential_store):
    """The fix must not turn a real deletion into an error, and the deleted credential must
    stop being served from the in-memory list the proxy routes on."""
    stored = CredentialItem(
        credential_name="doomed",
        credential_values={"api_key": "sk-old"},
        credential_info={"custom_llm_provider": "openai"},
    )
    survivor = CredentialItem(
        credential_name="keeper",
        credential_values={"api_key": "sk-keep"},
        credential_info={},
    )
    credential_store(in_memory=(stored, survivor), delete_by_name=AsyncMock(return_value=MagicMock()))

    response = _delete_credential("doomed")

    assert response.status_code == 200, response.text
    assert response.json()["success"] is True
    assert [credential.credential_name for credential in litellm.credential_list] == ["keeper"]


def test_delete_credential_leaves_a_credential_that_only_exists_in_memory_in_place(credential_store):
    """A credential declared in the config yaml is never written to the table, so the delete
    matches no row. Reporting success would be the same lie: it comes straight back on the next
    proxy boot."""
    config_only = CredentialItem(
        credential_name="from-config-yaml",
        credential_values={"api_key": "sk-config"},
        credential_info={},
        source="config",
    )
    credential_store(in_memory=(config_only,), delete_by_name=AsyncMock(return_value=None))

    response = _delete_credential("from-config-yaml")

    assert response.status_code == 405, response.text
    assert response.headers["allow"] == "GET"
    assert "defined in config" in response.json()["error"]["message"]
    assert [credential.credential_name for credential in litellm.credential_list] == ["from-config-yaml"]


def test_delete_credential_answers_500_when_the_database_is_not_connected(credential_store):
    """The handler used to ``return handle_exception_on_proxy(e)``, which makes the exception the
    response body and lets FastAPI answer 200. A DB-less proxy answered its own 500 as a success."""
    credential_store(connected=False)

    response = _delete_credential("any-name")

    assert response.status_code == 500, f"rejected delete answered {response.status_code}: {response.text}"


class _CredentialThatCannotBeMasked:
    """Stands in for anything that fails while ``GET /credentials`` builds its response."""

    credential_name = "unreadable"
    credential_info: dict = {}

    @property
    def credential_values(self):
        raise RuntimeError("credential store unreadable")


def test_get_credentials_answers_an_error_status_when_the_listing_fails(credential_store):
    """Same ``return`` instead of ``raise`` on the list route: a failed listing was serialized as
    a 200 whose body happened to be an error, so a caller reading the status saw an empty success."""
    credential_store(in_memory=(_CredentialThatCannotBeMasked(),))

    response = _list_credentials()

    assert response.status_code == 500, f"failed listing answered {response.status_code}: {response.text}"
    assert response.json().get("success") is not True


def _create_credential(body: dict):
    return _call_as("POST", "/credentials", body)


class _UniqueViolation(Exception):
    code = "P2002"


def test_create_credential_answers_409_when_the_name_is_already_taken(credential_store):
    """Regression: the unique index used to surface as a Prisma 500 that callers string-matched."""
    credential_store(
        create=AsyncMock(side_effect=_UniqueViolation("Unique constraint failed on the fields: (`credential_name`)")),
    )

    response = _create_credential(
        {"credential_name": "aws_bedrock", "credential_values": {"aws_access_key_id": "new"}, "credential_info": {}},
    )

    assert response.status_code == 409, f"name collision answered {response.status_code}: {response.text}"
    message = response.json()["error"]["message"]
    assert message == (
        "Credential 'aws_bedrock' already exists. Update it with PATCH /credentials/aws_bedrock, or delete it first."
    ), f"the operator reads this message verbatim: {message}"
    assert "Unique constraint" not in response.text, f"the Prisma internals must not leak: {response.text}"


def test_create_credential_still_answers_500_when_the_write_fails_for_another_reason(credential_store):
    credential_store(create=AsyncMock(side_effect=Exception("connection reset by peer")))

    response = _create_credential(
        {"credential_name": "aws_bedrock", "credential_values": {"aws_access_key_id": "new"}, "credential_info": {}},
    )

    assert response.status_code == 500, f"database fault answered {response.status_code}: {response.text}"


def test_create_credential_still_answers_200_for_a_name_that_is_free(credential_store):
    find_by_name = AsyncMock()
    credential_store(find_by_name=find_by_name, create=AsyncMock(return_value=None))

    response = _create_credential(
        {"credential_name": "brand_new", "credential_values": {"aws_access_key_id": "new"}, "credential_info": {}},
    )

    assert response.status_code == 200, response.text
    assert response.json()["success"] is True
    find_by_name.assert_not_awaited(), "the unique index is the guard; create must not add a lookup"


def test_update_credential_resolves_credential_values_from_model_id_like_create(credential_store):
    """Regression: PATCH dropped ``model_id`` from the body, so an update that named a
    deployment instead of raw values wrote whatever the caller sent, or nothing."""
    stored = CredentialItem(
        credential_name="from-deployment",
        credential_values={"api_key": "sk-old"},
        credential_info={},
    )
    update_by_name = AsyncMock(return_value=None)
    router = MagicMock()
    router.get_deployment.return_value = {"model_name": "gpt-5.2"}
    router.get_deployment_credentials.return_value = {"api_key": "sk-from-deployment"}
    credential_store(find_by_name=AsyncMock(return_value=stored), update_by_name=update_by_name, llm_router=router)

    response = _patch_credential(
        "from-deployment",
        {"credential_name": "from-deployment", "model_id": "deployment-1", "credential_info": {}},
    )

    assert response.status_code == 200, response.text
    router.get_deployment_credentials.assert_called_once_with("deployment-1")
    written = json.loads(update_by_name.await_args.kwargs["data"]["credential_values"])
    assert set(written) == {"api_key"}
    assert written["api_key"] != "sk-old", "the deployment's values must replace the stored ones"
    assert written["api_key"] != "sk-from-deployment", "values are encrypted before they reach the table"


def test_update_credential_answers_404_when_model_id_names_no_deployment(credential_store):
    stored = CredentialItem(
        credential_name="from-deployment", credential_values={"api_key": "sk-old"}, credential_info={}
    )
    update_by_name = AsyncMock(return_value=None)
    router = MagicMock()
    router.get_deployment.return_value = None
    credential_store(find_by_name=AsyncMock(return_value=stored), update_by_name=update_by_name, llm_router=router)

    response = _patch_credential(
        "from-deployment",
        {"credential_name": "from-deployment", "model_id": "no-such-deployment", "credential_info": {}},
    )

    assert response.status_code == 404, response.text
    update_by_name.assert_not_awaited()


def test_update_credential_answers_500_when_model_id_is_given_but_no_router_is_loaded(credential_store):
    stored = CredentialItem(
        credential_name="from-deployment", credential_values={"api_key": "sk-old"}, credential_info={}
    )
    update_by_name = AsyncMock(return_value=None)
    credential_store(find_by_name=AsyncMock(return_value=stored), update_by_name=update_by_name, llm_router=None)

    response = _patch_credential(
        "from-deployment",
        {"credential_name": "from-deployment", "model_id": "deployment-1", "credential_info": {}},
    )

    assert response.status_code == 500, response.text
    update_by_name.assert_not_awaited()


def test_update_credential_still_accepts_a_body_without_credential_values(credential_store):
    """Renaming or re-tagging a credential sends only ``credential_info``; that must not 422."""
    stored = CredentialItem(credential_name="existing", credential_values={"api_key": "sk-old"}, credential_info={})
    update_by_name = AsyncMock(return_value=None)
    credential_store(find_by_name=AsyncMock(return_value=stored), update_by_name=update_by_name)

    response = _patch_credential(
        "existing",
        {"credential_name": "existing", "credential_info": {"custom_llm_provider": "openai"}},
    )

    assert response.status_code == 200, response.text
    written = update_by_name.await_args.kwargs["data"]
    assert json.loads(written["credential_info"]) == {"custom_llm_provider": "openai"}
    assert set(json.loads(written["credential_values"])) == {"api_key"}, "stored values survive an info-only patch"


from contextlib import contextmanager as _ctx

from litellm.caching.dual_cache import DualCache
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
from litellm.proxy.credential_endpoints import user_provider_credentials as upc

_DEVICE_CODE_URL = "https://github.com/login/device/code"
_ACCESS_TOKEN_URL = "https://github.com/login/oauth/access_token"
_COPILOT_TOKEN_URL = "https://api.github.com/copilot_internal/v2/token"
_GITHUB_USER_URL = "https://api.github.com/user"


def _as_user(user_id="user-a"):
    return lambda: UserAPIKeyAuth(api_key="test-key", user_role="internal_user", user_id=user_id)


def _per_user_credential(name="copilot-cred"):
    return CredentialItem(
        credential_name=name,
        credential_values={"github_copilot_auth_type": "per_user_oauth"},
        credential_info={},
    )


def _connection_row(user_id="user-a", credential_name="copilot-cred", login="octo"):
    from datetime import datetime, timezone
    from types import SimpleNamespace

    return SimpleNamespace(
        user_id=user_id,
        credential_name=credential_name,
        provider="github_copilot",
        credential_b64=upc._encode(
            upc.GithubCopilotUserConnectionPayload(access_token="gho_secret", github_login=login)
        ),
        created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
        updated_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
    )


@pytest.fixture(autouse=True)
def _connection_master_key(monkeypatch):
    monkeypatch.setattr("litellm.proxy.proxy_server.master_key", "sk-test-master")


@_ctx
def _github_http(mapping):
    """Patch the shared async HTTP client to an httpx MockTransport that answers
    GitHub endpoints from {url: response-json dict or callable}; anything else fails."""

    def respond(request):
        entry = mapping.get(str(request.url))
        if entry is None:
            raise AssertionError(f"unexpected request to {request.url}")
        payload = entry(request) if callable(entry) else entry
        return httpx.Response(200, json=payload, request=request)

    client = AsyncHTTPHandler(transport=httpx.MockTransport(respond))
    with patch("litellm.llms.custom_httpx.http_handler.get_async_httpx_client", return_value=client):
        yield


_DEVICE_FLOW_START = {
    "device_code": "dc-1",
    "user_code": "UC-1",
    "verification_uri": "https://github.com/login/device",
    "expires_in": 900,
    "interval": 5,
}


def _patch_user_connection_env(monkeypatch, credentials, rows=()):
    """Per-user env: credential in memory, prisma table answering find_many with rows,
    and a real in-memory cache for device-flow + credential caching."""
    monkeypatch.setattr(litellm, "credential_list", list(credentials))
    prisma_client = MagicMock()
    table = MagicMock()
    table.find_many = AsyncMock(return_value=list(rows))
    table.find_unique = AsyncMock(return_value=rows[0] if rows else None)
    table.upsert = AsyncMock(return_value=None)
    table.delete_many = AsyncMock(return_value=None)
    prisma_client.db.litellm_userprovidercredentials = table
    prisma_client.writer_db = prisma_client.db
    monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", prisma_client)
    cache = DualCache()
    monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", cache)
    return table


def test_user_connections_lists_per_user_credentials_with_connection_state(monkeypatch):
    _patch_user_connection_env(
        monkeypatch,
        [_per_user_credential(), CredentialItem(credential_name="shared", credential_values={}, credential_info={})],
        rows=[_connection_row()],
    )
    response = _call_as("GET", "/credentials/user_connections", auth=_as_user())
    assert response.status_code == 200, response.text
    assert response.json() == {
        "connections": [
            {
                "credential_name": "copilot-cred",
                "provider": "github_copilot",
                "connected": True,
                "github_login": "octo",
                "connected_at": "2026-01-01T00:00:00+00:00",
            }
        ]
    }


def test_user_connections_requires_a_user_id(monkeypatch):
    _patch_user_connection_env(monkeypatch, [_per_user_credential()])
    response = _call_as("GET", "/credentials/user_connections", auth=_as_admin)
    assert response.status_code == 401


def test_user_connection_start_returns_device_flow_fields(monkeypatch):
    _patch_user_connection_env(monkeypatch, [_per_user_credential()])
    with _github_http({_DEVICE_CODE_URL: _DEVICE_FLOW_START}):
        response = _call_as("POST", "/credentials/copilot-cred/user_connection/start", auth=_as_user())
    assert response.status_code == 200, response.text
    assert {key: response.json()[key] for key in ("user_code", "verification_uri", "expires_in", "interval")} == {
        "user_code": "UC-1",
        "verification_uri": "https://github.com/login/device",
        "expires_in": 900,
        "interval": 5,
    }
    handle = response.json()["flow_handle"]
    assert isinstance(handle, str) and "device_code" not in handle and _DEVICE_FLOW_START["device_code"] not in handle


def test_user_connection_start_404s_for_shared_credential(monkeypatch):
    _patch_user_connection_env(
        monkeypatch, [CredentialItem(credential_name="shared", credential_values={}, credential_info={})]
    )
    response = _call_as("POST", "/credentials/shared/user_connection/start", auth=_as_user())
    assert response.status_code == 404


def test_user_connection_poll_rejects_an_undecryptable_flow_handle(monkeypatch):
    _patch_user_connection_env(monkeypatch, [_per_user_credential()])
    response = _call_as(
        "POST",
        "/credentials/copilot-cred/user_connection/poll",
        json_body={"flow_handle": "not-a-real-handle"},
        auth=_as_user(),
    )
    assert response.status_code == 400


def test_user_connection_poll_connected_persists_and_reports_login(monkeypatch):
    table = _patch_user_connection_env(monkeypatch, [_per_user_credential()])
    with _github_http(
        {
            _DEVICE_CODE_URL: _DEVICE_FLOW_START,
            _ACCESS_TOKEN_URL: {"access_token": "gho_1"},
            _COPILOT_TOKEN_URL: {"token": "copilot-tok", "expires_at": 4102444800},
            _GITHUB_USER_URL: {"login": "octo"},
        }
    ):
        start = _call_as("POST", "/credentials/copilot-cred/user_connection/start", auth=_as_user())
        assert start.status_code == 200, start.text
        poll = _call_as(
            "POST",
            "/credentials/copilot-cred/user_connection/poll",
            json_body={"flow_handle": start.json()["flow_handle"]},
            auth=_as_user(),
        )
    assert poll.status_code == 200, poll.text
    assert poll.json() == {"status": "connected", "interval": None, "github_login": "octo"}
    table.upsert.assert_awaited_once()


def test_user_connection_poll_pending(monkeypatch):
    table = _patch_user_connection_env(monkeypatch, [_per_user_credential()])
    with _github_http(
        {
            _DEVICE_CODE_URL: _DEVICE_FLOW_START,
            _ACCESS_TOKEN_URL: {"error": "authorization_pending"},
        }
    ):
        start = _call_as("POST", "/credentials/copilot-cred/user_connection/start", auth=_as_user())
        poll = _call_as(
            "POST",
            "/credentials/copilot-cred/user_connection/poll",
            json_body={"flow_handle": start.json()["flow_handle"]},
            auth=_as_user(),
        )
    assert poll.json()["status"] == "pending"
    table.upsert.assert_not_awaited()


def test_delete_user_connection_disconnects_and_is_idempotent(monkeypatch):
    table = _patch_user_connection_env(monkeypatch, [_per_user_credential()], rows=[_connection_row()])
    response = _call_as("DELETE", "/credentials/copilot-cred/user_connection", auth=_as_user())
    assert response.status_code == 200
    assert response.json() == {"status": "disconnected"}
    table.delete_many.assert_awaited_once_with(where={"user_id": "user-a", "credential_name": "copilot-cred"})

    # idempotent: row now gone
    table.find_unique = AsyncMock(return_value=None)
    again = _call_as("DELETE", "/credentials/copilot-cred/user_connection", auth=_as_user())
    assert again.status_code == 200
    assert again.json() == {"status": "disconnected"}


def test_deleting_the_credential_purges_user_connections(monkeypatch, credential_store):
    rows = [_connection_row(user_id="u1"), _connection_row(user_id="u2")]
    credential_store(
        in_memory=[_per_user_credential()],
        delete_by_name=AsyncMock(return_value=_per_user_credential()),
    )
    prisma_client = MagicMock()
    table = MagicMock()
    table.find_many = AsyncMock(return_value=rows)
    table.delete_many = AsyncMock(return_value=None)
    prisma_client.db.litellm_userprovidercredentials = table
    prisma_client.writer_db = prisma_client.db
    prisma_client.db.litellm_credentialstable.find_unique = AsyncMock(return_value=None)
    monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", prisma_client)
    monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", DualCache())

    response = _delete_credential("copilot-cred")
    assert response.status_code == 200, response.text
    table.delete_many.assert_awaited_with(where={"credential_name": "copilot-cred"})


def test_user_connections_isolated_per_user(monkeypatch):
    """Two users connected to the same credential: A's listing shows only A's login,
    A's delete removes only A's row, and a poll only reads A's device-flow entry."""

    rows = [_connection_row(user_id="user-a", login="octo"), _connection_row(user_id="user-b", login="hubot")]
    table = _patch_user_connection_env(monkeypatch, [_per_user_credential()], rows=rows)

    async def _find_many_matching_user(*, where):
        return [row for row in rows if row.user_id == where["user_id"]]

    table.find_many = AsyncMock(side_effect=_find_many_matching_user)

    listing = _call_as("GET", "/credentials/user_connections", auth=_as_user("user-a"))
    assert listing.status_code == 200
    body = listing.json()["connections"]
    assert [c["github_login"] for c in body] == ["octo"]
    # find_many must be scoped to the calling user
    where = table.find_many.await_args.kwargs["where"]
    assert where["user_id"] == "user-a"

    # B's in-flight device flow handle is bound to B: A polling with it is rejected
    poll = _call_as(
        "POST",
        "/credentials/copilot-cred/user_connection/poll",
        json_body={"flow_handle": _flow_handle_for(user_id="user-b")},
        auth=_as_user("user-a"),
    )
    assert poll.status_code == 400
    table.upsert.assert_not_awaited()

    delete = _call_as("DELETE", "/credentials/copilot-cred/user_connection", auth=_as_user("user-a"))
    assert delete.json() == {"status": "disconnected"}
    table.delete_many.assert_awaited_once_with(where={"user_id": "user-a", "credential_name": "copilot-cred"})


@pytest.mark.parametrize("seat_status", [403, 404])
def test_user_connection_poll_reports_no_copilot_seat(monkeypatch, seat_status):
    """A GitHub seat check that rejects the user's token maps to no_copilot_seat, stores
    nothing, and clears the in-flight device code."""
    table = _patch_user_connection_env(monkeypatch, [_per_user_credential()])

    def respond(request):
        url = str(request.url)
        if url == _DEVICE_CODE_URL:
            return httpx.Response(200, json=_DEVICE_FLOW_START, request=request)
        if url == _ACCESS_TOKEN_URL:
            return httpx.Response(200, json={"access_token": "gho_seat"}, request=request)
        if url == _COPILOT_TOKEN_URL:
            return httpx.Response(seat_status, json={}, request=request)
        pytest.fail(f"unexpected request to {url}")

    client_obj = AsyncHTTPHandler(transport=httpx.MockTransport(respond))
    with patch("litellm.llms.custom_httpx.http_handler.get_async_httpx_client", return_value=client_obj):
        start = _call_as("POST", "/credentials/copilot-cred/user_connection/start", auth=_as_user())
        assert start.status_code == 200
        poll = _call_as(
            "POST",
            "/credentials/copilot-cred/user_connection/poll",
            json_body={"flow_handle": start.json()["flow_handle"]},
            auth=_as_user(),
        )
    assert poll.status_code == 200, poll.text
    assert poll.json()["status"] == "no_copilot_seat"
    table.upsert.assert_not_awaited()


def test_user_connection_poll_rate_limit_returns_429(monkeypatch):
    table = _patch_user_connection_env(monkeypatch, [_per_user_credential()])

    def respond(request):
        url = str(request.url)
        if url == _DEVICE_CODE_URL:
            return httpx.Response(200, json=_DEVICE_FLOW_START, request=request)
        if url == _ACCESS_TOKEN_URL:
            return httpx.Response(200, json={"access_token": "gho_limited"}, request=request)
        if url == _COPILOT_TOKEN_URL:
            return httpx.Response(429, json={}, request=request)
        pytest.fail(f"unexpected request to {url}")

    client_obj = AsyncHTTPHandler(transport=httpx.MockTransport(respond))
    with patch("litellm.llms.custom_httpx.http_handler.get_async_httpx_client", return_value=client_obj):
        start = _call_as("POST", "/credentials/copilot-cred/user_connection/start", auth=_as_user())
        poll = _call_as(
            "POST",
            "/credentials/copilot-cred/user_connection/poll",
            json_body={"flow_handle": start.json()["flow_handle"]},
            auth=_as_user(),
        )
    assert poll.status_code == 429
    table.upsert.assert_not_awaited()


@pytest.mark.parametrize(
    "method,path",
    [
        ("POST", "/credentials/missing/user_connection/start"),
        ("POST", "/credentials/missing/user_connection/poll"),
        ("DELETE", "/credentials/missing/user_connection"),
        ("POST", "/credentials/shared/user_connection/poll"),
        ("DELETE", "/credentials/shared/user_connection"),
    ],
)
def test_user_connection_routes_404_for_missing_or_shared_credentials(monkeypatch, method, path):
    _patch_user_connection_env(
        monkeypatch, [CredentialItem(credential_name="shared", credential_values={}, credential_info={})]
    )
    response = _call_as(method, path, json_body={"flow_handle": "x"}, auth=_as_user())
    assert response.status_code == 404


@pytest.mark.parametrize(
    "method,path",
    [
        ("GET", "/credentials/user_connections"),
        ("POST", "/credentials/copilot-cred/user_connection/start"),
        ("POST", "/credentials/copilot-cred/user_connection/poll"),
        ("DELETE", "/credentials/copilot-cred/user_connection"),
    ],
)
def test_user_connection_routes_401_without_user_id(monkeypatch, method, path):
    _patch_user_connection_env(monkeypatch, [_per_user_credential()])
    response = _call_as(method, path, json_body={"flow_handle": "x"}, auth=_as_admin)  # admin token has no user_id
    assert response.status_code == 401


def test_user_connection_routes_roles():
    """internal_user is allowed (self-scoped), internal_user_view_only is denied, proxy_admin
    is allowed; and POST /credentials stays admin-only for an internal user."""
    from litellm.proxy.auth.route_checks import RouteChecks

    routes = [
        ("/credentials/user_connections", "GET"),
        ("/credentials/copilot-cred/user_connection/start", "POST"),
        ("/credentials/copilot-cred/user_connection/poll", "POST"),
        ("/credentials/copilot-cred/user_connection", "DELETE"),
    ]

    def outcome(role, route, method="GET"):
        if role == LitellmUserRoles.PROXY_ADMIN:
            return "allowed"
        user_obj = LiteLLM_UserTable(user_id="u", user_email="u@x", user_role=role.value)
        valid_token = UserAPIKeyAuth(user_id="u", user_role=role)
        request = MagicMock(spec=Request)
        request.method = method
        request.query_params = {}
        try:
            RouteChecks.non_proxy_admin_allowed_routes_check(
                user_obj=user_obj,
                _user_role=role.value,
                route=route,
                request=request,
                valid_token=valid_token,
                request_data={},
            )
        except HTTPException as exc:
            return f"denied:{exc.status_code}"
        except Exception:
            return "denied"
        return "allowed"

    for route, method in routes:
        assert outcome(LitellmUserRoles.INTERNAL_USER, route, method) == "allowed"
        assert outcome(LitellmUserRoles.INTERNAL_USER_VIEW_ONLY, route, method) == "denied"
    assert outcome(LitellmUserRoles.INTERNAL_USER, "/credentials", method="POST") != "allowed"


class TestNonAdminCannotSetPerUserOauthOnCredential:
    """github_copilot_auth_type selects the GitHub OAuth path; it is server-owned WIF, so a
    team admin must not set it on a credential and a proxy admin can."""

    def test_non_admin_cannot_create_a_per_user_oauth_credential(self):
        with patch(  # test-quality-ok: the proxy wiring under test is what this patches
            "litellm.proxy.proxy_server.prisma_client", MagicMock()
        ):
            response = _post_credential(
                {
                    "credential_name": "attacker-cred",
                    "credential_values": {"github_copilot_auth_type": "per_user_oauth"},
                    "credential_info": {"custom_llm_provider": "github_copilot"},
                },
                auth=_as_non_admin,
            )
        assert response.status_code == 403, response.text
        assert "github_copilot_auth_type" in response.json()["error"]["message"]

    def test_non_admin_cannot_patch_a_credential_to_per_user_oauth(self, restore_credential_list):
        stored = CredentialItem(
            credential_name="existing",
            credential_values={"api_key": "sk"},
            credential_info={"custom_llm_provider": "github_copilot"},
        )
        with _repository_holding(stored) as repository:
            repository.find_unique_by_name = AsyncMock(return_value=stored)
            response = _patch_credential(
                "existing",
                {
                    "credential_name": "existing",
                    "credential_values": {"github_copilot_auth_type": "per_user_oauth"},
                    "credential_info": {},
                },
                auth=_as_non_admin,
            )
        assert response.status_code == 403, response.text

    def test_proxy_admin_can_create_a_per_user_oauth_credential(self, restore_credential_list):
        with _repository_holding(None) as repository:
            response = _post_credential(
                {
                    "credential_name": "admin-cred",
                    "credential_values": {"github_copilot_auth_type": "per_user_oauth"},
                    "credential_info": {"custom_llm_provider": "github_copilot"},
                },
            )
        assert response.status_code == 200, response.text
        repository.create.assert_awaited_once()

    def test_proxy_admin_can_patch_a_credential_to_per_user_oauth(self, restore_credential_list):
        stored = CredentialItem(
            credential_name="existing",
            credential_values={"api_key": "sk"},
            credential_info={"custom_llm_provider": "github_copilot"},
        )
        with _repository_holding(stored) as repository:
            repository.find_unique_by_name = AsyncMock(return_value=stored)
            repository.update_by_name = AsyncMock(return_value=stored)
            response = _patch_credential(
                "existing",
                {
                    "credential_name": "existing",
                    "credential_values": {"github_copilot_auth_type": "per_user_oauth"},
                    "credential_info": {},
                },
            )
        assert response.status_code == 200, response.text


def test_label_only_patch_does_not_purge_user_connections():
    """A PATCH that only changes display_name sends no credential_name in the
    body, and that must not read as a rename that purges every user's stored
    connection."""
    stored: Final = CredentialItem(
        credential_name="copilot-cred",
        credential_values={"github_copilot_auth_type": "per_user_oauth"},
        credential_info={"custom_llm_provider": "github_copilot"},
    )
    with _repository_holding(stored):
        from litellm.proxy import proxy_server

        table: Final = MagicMock()
        table.find_many = AsyncMock(return_value=[_connection_row()])
        table.delete_many = AsyncMock(return_value=None)
        proxy_server.prisma_client.db.litellm_userprovidercredentials = table
        proxy_server.prisma_client.writer_db = proxy_server.prisma_client.db

        response: Final = _patch_credential(
            "copilot-cred",
            {"display_name": "Renamed Copilot", "credential_info": {"custom_llm_provider": "github_copilot"}},
        )

    assert response.status_code == 200, response.text
    table.find_many.assert_not_awaited()
    table.delete_many.assert_not_awaited()


def test_switching_away_from_per_user_oauth_purges_user_connections():
    stored: Final = CredentialItem(
        credential_name="copilot-cred",
        credential_values={"github_copilot_auth_type": "per_user_oauth"},
        credential_info={"custom_llm_provider": "github_copilot"},
    )
    with _repository_holding(stored):
        from litellm.proxy import proxy_server

        table: Final = MagicMock()
        table.find_many = AsyncMock(return_value=[_connection_row()])
        table.delete_many = AsyncMock(return_value=None)
        proxy_server.prisma_client.db.litellm_userprovidercredentials = table
        proxy_server.prisma_client.writer_db = proxy_server.prisma_client.db

        response: Final = _patch_credential(
            "copilot-cred",
            {
                "credential_name": "copilot-cred",
                "credential_values": {},
                "credential_values_to_delete": ["github_copilot_auth_type"],
                "credential_info": {"custom_llm_provider": "github_copilot"},
            },
        )

    assert response.status_code == 200, response.text
    table.delete_many.assert_awaited_with(where={"credential_name": "copilot-cred"})


def test_pending_device_flow_survives_disconnect_via_stateless_handle(monkeypatch):
    """The device flow is stateless: no worker-local entry exists to clear on
    disconnect, so a still-valid issued handle polled on any worker completes."""
    _patch_user_connection_env(monkeypatch, [_per_user_credential()], rows=[_connection_row()])

    with _github_http({_DEVICE_CODE_URL: _DEVICE_FLOW_START}):
        start = _call_as("POST", "/credentials/copilot-cred/user_connection/start", auth=_as_user())
        assert start.status_code == 200
    delete = _call_as("DELETE", "/credentials/copilot-cred/user_connection", auth=_as_user())
    assert delete.json() == {"status": "disconnected"}


def test_user_connection_cache_keys_cannot_collide_on_colons():
    """user_id/credential_name are joined losslessly, so ('a:b','c') and ('a','b:c')
    can never share a cache or device-flow entry."""
    assert upc._cache_key("a:b", "c") != upc._cache_key("a", "b:c")


def test_user_connection_slashed_credential_name_routes(monkeypatch):
    """A credential named 'team/copilot' must still reach start/poll/delete: the
    routes take :path parameters the same way the admin CRUD routes do."""
    table = _patch_user_connection_env(monkeypatch, [_per_user_credential(name="team/copilot")])
    with _github_http(
        {
            _DEVICE_CODE_URL: _DEVICE_FLOW_START,
            _ACCESS_TOKEN_URL: {"error": "authorization_pending"},
        }
    ):
        start = _call_as("POST", "/credentials/team/copilot/user_connection/start", auth=_as_user())
        assert start.status_code == 200, start.text
        assert start.json()["user_code"] == "UC-1"

        poll = _call_as(
            "POST",
            "/credentials/team/copilot/user_connection/poll",
            json_body={"flow_handle": start.json()["flow_handle"]},
            auth=_as_user(),
        )
    assert poll.status_code == 200
    assert poll.json()["status"] == "pending"

    delete = _call_as("DELETE", "/credentials/team/copilot/user_connection", auth=_as_user())
    assert delete.status_code == 200
    assert delete.json() == {"status": "disconnected"}
    table.delete_many.assert_awaited_once_with(where={"user_id": "user-a", "credential_name": "team/copilot"})


def test_user_connection_route_check_allows_slashed_names_for_internal_users():
    """The RouteChecks allowlist entries use :path placeholders too, so an internal
    user's request to /credentials/team/copilot/user_connection/... still matches."""
    from litellm.proxy.auth.route_checks import RouteChecks

    for route in (
        "/credentials/team/copilot/user_connection/start",
        "/credentials/team/copilot/user_connection/poll",
        "/credentials/team/copilot/user_connection",
    ):
        assert RouteChecks._route_matches_pattern(
            route=route, pattern=route.replace("team/copilot", "{credential_name:path}")
        )

    user_obj = LiteLLM_UserTable(user_id="u", user_email="u@x", user_role=LitellmUserRoles.INTERNAL_USER.value)
    request = MagicMock(spec=Request)
    request.method = "POST"
    request.query_params = {}
    RouteChecks.non_proxy_admin_allowed_routes_check(
        user_obj=user_obj,
        _user_role=LitellmUserRoles.INTERNAL_USER.value,
        route="/credentials/team/copilot/user_connection/poll",
        request=request,
        valid_token=UserAPIKeyAuth(user_id="u", user_role=LitellmUserRoles.INTERNAL_USER),
        request_data={},
    )


def _flow_handle_for(user_id="user-a", credential_name="copilot-cred", expires_at=None):
    import time as _time

    from litellm.models.credentials import UserConnectionFlowHandle
    from litellm.proxy.common_utils.encrypt_decrypt_utils import encrypt_value_helper

    return encrypt_value_helper(
        UserConnectionFlowHandle(
            user_id=user_id,
            credential_name=credential_name,
            device_code="DC-1",
            interval=5,
            expires_at=expires_at if expires_at is not None else _time.time() + 900,
        ).model_dump_json()
    )


def test_user_connection_poll_rejects_a_handle_bound_to_another_user(monkeypatch):
    _patch_user_connection_env(monkeypatch, [_per_user_credential()])
    response = _call_as(
        "POST",
        "/credentials/copilot-cred/user_connection/poll",
        json_body={"flow_handle": _flow_handle_for(user_id="user-b")},
        auth=_as_user("user-a"),
    )
    assert response.status_code == 400


def test_user_connection_poll_rejects_a_handle_bound_to_another_credential(monkeypatch):
    table = _patch_user_connection_env(monkeypatch, [_per_user_credential()])
    with _github_http({_DEVICE_CODE_URL: _DEVICE_FLOW_START}):
        start = _call_as("POST", "/credentials/copilot-cred/user_connection/start", auth=_as_user())
        assert start.status_code == 200
    forged = _flow_handle_for(credential_name="other-cred")
    response = _call_as(
        "POST",
        "/credentials/copilot-cred/user_connection/poll",
        json_body={"flow_handle": forged},
        auth=_as_user(),
    )
    assert response.status_code == 400
    assert start.json()["flow_handle"] != forged
    table.upsert.assert_not_awaited()


def test_user_connection_poll_rejects_an_expired_handle(monkeypatch):
    import time as _time

    _patch_user_connection_env(monkeypatch, [_per_user_credential()])
    response = _call_as(
        "POST",
        "/credentials/copilot-cred/user_connection/poll",
        json_body={"flow_handle": _flow_handle_for(expires_at=_time.time() - 1)},
        auth=_as_user(),
    )
    assert response.status_code == 400


def test_user_connection_poll_valid_handle_works_with_a_fresh_cache(monkeypatch):
    """The handle carries the device code: a poll on a worker whose cache is empty
    still completes the flow, no cross-worker state required."""
    table = _patch_user_connection_env(monkeypatch, [_per_user_credential()])
    with _github_http(
        {
            _DEVICE_CODE_URL: _DEVICE_FLOW_START,
            _ACCESS_TOKEN_URL: {"access_token": "gho_1"},
            _COPILOT_TOKEN_URL: {"token": "copilot-tok", "expires_at": 4102444800},
            _GITHUB_USER_URL: {"login": "octo"},
        }
    ):
        start = _call_as("POST", "/credentials/copilot-cred/user_connection/start", auth=_as_user())
        assert start.status_code == 200, start.text
        # simulate a different worker: swap in a brand new empty cache
        from litellm.caching.dual_cache import DualCache
        from litellm.proxy import proxy_server

        monkeypatch.setattr(proxy_server, "user_api_key_cache", DualCache())
        poll = _call_as(
            "POST",
            "/credentials/copilot-cred/user_connection/poll",
            json_body={"flow_handle": start.json()["flow_handle"]},
            auth=_as_user(),
        )
    assert poll.status_code == 200, poll.text
    assert poll.json()["status"] == "connected"
    table.upsert.assert_awaited_once()


def test_delete_user_connection_fails_closed_when_the_tombstone_write_fails(monkeypatch):
    """Redis attached but its set() raises: the disconnect must not delete the row,
    or the cached token would keep working until TTL with nothing left to revoke."""
    table = _patch_user_connection_env(monkeypatch, [_per_user_credential()], rows=[_connection_row()])
    from types import SimpleNamespace

    from litellm.proxy import proxy_server

    failing_redis = SimpleNamespace(
        async_set_cache=AsyncMock(side_effect=ConnectionError("redis down")),
        async_get_cache=AsyncMock(return_value=None),
        async_delete_cache=AsyncMock(return_value=None),
    )
    monkeypatch.setattr(proxy_server, "user_api_key_cache", DualCache(redis_cache=failing_redis))

    response = _call_as("DELETE", "/credentials/copilot-cred/user_connection", auth=_as_user())
    assert response.status_code == 503, response.text
    table.delete_many.assert_not_awaited()


def test_delete_user_connection_fails_when_the_tombstone_write_silently_noops(monkeypatch):
    """RedisCache.async_set_cache swallows client errors, so a set() that returns
    without writing still looks successful. The disconnect must verify the
    tombstone by reading the key back and refuse to delete the row."""
    table = _patch_user_connection_env(monkeypatch, [_per_user_credential()], rows=[_connection_row()])
    from types import SimpleNamespace

    from litellm.proxy import proxy_server

    store: dict = {}
    silent_redis = SimpleNamespace(
        async_set_cache=AsyncMock(return_value=None),  # reports success, writes nothing
        async_get_cache=AsyncMock(side_effect=lambda key: store.get(key)),
        async_delete_cache=AsyncMock(return_value=None),
    )
    monkeypatch.setattr(proxy_server, "user_api_key_cache", DualCache(redis_cache=silent_redis))

    response = _call_as("DELETE", "/credentials/copilot-cred/user_connection", auth=_as_user())
    assert response.status_code == 503, response.text
    table.delete_many.assert_not_awaited()


def test_user_connection_poll_reports_503_when_the_cache_overwrite_cannot_land(monkeypatch):
    """The connect save is durable, so a stale not-connected marker that survives
    both the overwrite and the delete must surface as a retryable 503 rather
    than a false "connected"."""
    table = _patch_user_connection_env(monkeypatch, [_per_user_credential()])
    from types import SimpleNamespace

    from litellm.proxy import proxy_server

    stale_redis = SimpleNamespace(
        async_set_cache=AsyncMock(return_value=None),
        async_get_cache=AsyncMock(return_value="stale-value"),
        async_delete_cache=AsyncMock(return_value=None),
    )
    monkeypatch.setattr(proxy_server, "user_api_key_cache", DualCache(redis_cache=stale_redis))

    with _github_http(
        {
            _DEVICE_CODE_URL: _DEVICE_FLOW_START,
            _ACCESS_TOKEN_URL: {"access_token": "gho_1"},
            _COPILOT_TOKEN_URL: {"token": "copilot-tok", "expires_at": 4102444800},
            _GITHUB_USER_URL: {"login": "octo"},
        }
    ):
        start = _call_as("POST", "/credentials/copilot-cred/user_connection/start", auth=_as_user())
        poll = _call_as(
            "POST",
            "/credentials/copilot-cred/user_connection/poll",
            json_body={"flow_handle": start.json()["flow_handle"]},
            auth=_as_user(),
        )
    assert poll.status_code == 503, poll.text
    assert "cache could not be refreshed" in poll.text
    table.upsert.assert_awaited_once()


def _labeled_credential(name: str = "openai-prod", display_name: str | None = "Prod OpenAI") -> CredentialItem:
    return CredentialItem(
        credential_name=name,
        display_name=display_name,
        credential_values={"api_key": "sk-old"},
        credential_info={"custom_llm_provider": "openai"},
    )


class TestCredentialDisplayName:
    def test_create_stores_the_trimmed_display_name_in_the_row_and_in_memory(self, restore_credential_list):
        with _repository_holding(None) as repository:
            response = _post_credential(
                {
                    "credential_name": "openai-prod",
                    "display_name": "  Prod OpenAI  ",
                    "credential_values": {"api_key": "sk-new"},
                    "credential_info": {"custom_llm_provider": "openai"},
                }
            )

        assert response.status_code == 200, response.text
        assert repository.create.await_args.kwargs["data"]["display_name"] == "Prod OpenAI"
        assert [(c.credential_name, c.display_name) for c in litellm.credential_list] == [
            ("openai-prod", "Prod OpenAI")
        ]

    @pytest.mark.parametrize("display_name", ["", "   ", "x" * 256])
    def test_create_rejects_an_unusable_display_name(self, restore_credential_list, display_name):
        with _repository_holding(None) as repository:
            response = _post_credential(
                {
                    "credential_name": "openai-prod",
                    "display_name": display_name,
                    "credential_values": {"api_key": "sk-new"},
                    "credential_info": {},
                }
            )

        assert response.status_code == 400, response.text
        assert response.json()["error"]["param"] == "display_name"
        repository.create.assert_not_awaited()
        assert litellm.credential_list == []

    def test_create_accepts_a_display_name_of_exactly_the_maximum_length(self, restore_credential_list):
        with _repository_holding(None) as repository:
            response = _post_credential(
                {
                    "credential_name": "openai-prod",
                    "display_name": "x" * 255,
                    "credential_values": {"api_key": "sk-new"},
                    "credential_info": {},
                }
            )

        assert response.status_code == 200, response.text
        assert repository.create.await_args.kwargs["data"]["display_name"] == "x" * 255

    def test_patch_relabels_without_touching_the_name_or_the_values(self, restore_credential_list, monkeypatch):
        monkeypatch.setattr(litellm, "credential_list", [_labeled_credential()])
        with _repository_holding(_labeled_credential()) as repository:
            response = _patch_credential("openai-prod", {"display_name": "Staging OpenAI", "credential_info": {}})

        assert response.status_code == 200, response.text
        repository.update_by_name.assert_awaited_once()
        assert repository.update_by_name.await_args.args[0] == "openai-prod"
        written = repository.update_by_name.await_args.kwargs["data"]
        assert written["credential_name"] == "openai-prod"
        assert written["display_name"] == "Staging OpenAI"
        assert json.loads(written["credential_values"]) == {"api_key": "sk-old"}
        assert [(c.credential_name, c.display_name, c.credential_values) for c in litellm.credential_list] == [
            ("openai-prod", "Staging OpenAI", {"api_key": "sk-old"})
        ]

    def test_patch_without_display_name_keeps_the_stored_one(self, restore_credential_list, monkeypatch):
        monkeypatch.setattr(litellm, "credential_list", [_labeled_credential()])
        with _repository_holding(_labeled_credential()) as repository:
            response = _patch_credential("openai-prod", {"credential_info": {"description": "rotated"}})

        assert response.status_code == 200, response.text
        assert repository.update_by_name.await_args.kwargs["data"]["display_name"] == "Prod OpenAI"
        assert litellm.credential_list[0].display_name == "Prod OpenAI"

    def test_patch_with_null_display_name_clears_it(self, restore_credential_list, monkeypatch):
        monkeypatch.setattr(litellm, "credential_list", [_labeled_credential()])
        with _repository_holding(_labeled_credential()) as repository:
            response = _patch_credential("openai-prod", {"display_name": None, "credential_info": {}})

        assert response.status_code == 200, response.text
        written = repository.update_by_name.await_args.kwargs["data"]
        assert "display_name" in written
        assert written["display_name"] is None
        assert litellm.credential_list[0].display_name is None

    @pytest.mark.parametrize("display_name", ["", "   ", "x" * 256])
    def test_patch_rejects_an_unusable_display_name_and_changes_nothing(
        self, restore_credential_list, monkeypatch, display_name
    ):
        monkeypatch.setattr(litellm, "credential_list", [_labeled_credential()])
        with _repository_holding(_labeled_credential()) as repository:
            response = _patch_credential("openai-prod", {"display_name": display_name, "credential_info": {}})

        assert response.status_code == 400, response.text
        assert response.json()["error"]["param"] == "display_name"
        repository.update_by_name.assert_not_awaited()
        assert litellm.credential_list == [_labeled_credential()]

    def test_patch_stores_the_trimmed_display_name(self, restore_credential_list, monkeypatch):
        monkeypatch.setattr(litellm, "credential_list", [_labeled_credential()])
        with _repository_holding(_labeled_credential()) as repository:
            response = _patch_credential("openai-prod", {"display_name": "  Staging OpenAI  ", "credential_info": {}})

        assert response.status_code == 200, response.text
        assert repository.update_by_name.await_args.kwargs["data"]["display_name"] == "Staging OpenAI"
        assert litellm.credential_list[0].display_name == "Staging OpenAI"

    def test_patch_treats_an_empty_credential_name_as_omitted(self, restore_credential_list, monkeypatch):
        monkeypatch.setattr(litellm, "credential_list", [_labeled_credential()])
        with _repository_holding(_labeled_credential()) as repository:
            response = _patch_credential(
                "openai-prod", {"credential_name": "", "display_name": "Staging OpenAI", "credential_info": {}}
            )

        assert response.status_code == 200, response.text
        assert repository.update_by_name.await_args.args[0] == "openai-prod"
        assert repository.update_by_name.await_args.kwargs["data"]["credential_name"] == "openai-prod"

    def test_patch_rejects_a_different_credential_name_and_changes_nothing(self, restore_credential_list, monkeypatch):
        monkeypatch.setattr(litellm, "credential_list", [_labeled_credential()])
        with _repository_holding(_labeled_credential()) as repository:
            response = _patch_credential(
                "openai-prod",
                {
                    "credential_name": "openai-production",
                    "credential_values": {"api_key": "sk-new"},
                    "credential_info": {},
                },
            )

        assert response.status_code == 400, response.text
        assert response.json()["error"]["param"] == "credential_name"
        assert "display_name" in response.json()["error"]["message"]
        repository.update_by_name.assert_not_awaited()
        assert litellm.credential_list == [_labeled_credential()]

    def test_patch_accepts_the_same_credential_name_echoed_back(self, restore_credential_list, monkeypatch):
        monkeypatch.setattr(litellm, "credential_list", [_labeled_credential()])
        with _repository_holding(_labeled_credential()) as repository:
            response = _patch_credential(
                "openai-prod",
                {"credential_name": "openai-prod", "credential_values": {"api_key": "sk-new"}, "credential_info": {}},
            )

        assert response.status_code == 200, response.text
        assert [(c.credential_name, c.credential_values) for c in litellm.credential_list] == [
            ("openai-prod", {"api_key": "sk-new"})
        ]

    def test_patch_on_a_config_credential_is_rejected_as_config_owned(self, restore_credential_list, monkeypatch):
        config_credential = _labeled_credential(display_name=None).model_copy(update={"source": "config"})
        monkeypatch.setattr(litellm, "credential_list", [config_credential])
        with _repository_holding(None) as repository:
            response = _patch_credential("openai-prod", {"display_name": "Prod", "credential_info": {}})

        assert response.status_code == 405, response.text
        assert response.headers["allow"] == "GET"
        assert "defined in config" in response.json()["error"]["message"]
        repository.update_by_name.assert_not_awaited()
        assert litellm.credential_list == [config_credential]

    def test_patch_on_an_unknown_credential_still_answers_404(self, restore_credential_list):
        with _repository_holding(None):
            response = _patch_credential("nowhere", {"display_name": "Prod", "credential_info": {}})

        assert response.status_code == 404, response.text

    @pytest.mark.parametrize("method", ["PATCH", "DELETE"])
    def test_a_db_credential_already_gone_from_the_db_answers_404_not_config_owned(
        self, restore_credential_list, monkeypatch, method
    ):
        monkeypatch.setattr(litellm, "credential_list", [_labeled_credential()])
        with _repository_holding(None) as repository:
            response = (
                _patch_credential("openai-prod", {"display_name": "Prod", "credential_info": {}})
                if method == "PATCH"
                else _delete_credential("openai-prod")
            )

        assert response.status_code == 404, response.text
        repository.update_by_name.assert_not_awaited()

    def test_patch_on_a_name_stored_in_the_db_and_config_applies_to_the_row(self, restore_credential_list, monkeypatch):
        monkeypatch.setattr(litellm, "credential_list", [_labeled_credential(display_name=None)])
        with _repository_holding(_labeled_credential(display_name=None)) as repository:
            response = _patch_credential("openai-prod", {"display_name": "Prod", "credential_info": {}})

        assert response.status_code == 200, response.text
        assert repository.update_by_name.await_args.kwargs["data"]["display_name"] == "Prod"

    def test_reads_return_display_name_and_source(self, restore_credential_list, monkeypatch):
        config_credential = CredentialItem(
            credential_name="from-config",
            credential_values={"api_key": "sk-config-value"},
            credential_info={},
            source="config",
        )
        monkeypatch.setattr(litellm, "credential_list", [_labeled_credential(), config_credential])

        listed = {entry["credential_name"]: entry for entry in _list_credentials().json()["credentials"]}
        by_name = _call_as("GET", "/credentials/by_name/openai-prod").json()
        config_by_name = _call_as("GET", "/credentials/by_name/from-config").json()

        assert (listed["openai-prod"]["display_name"], listed["openai-prod"]["source"]) == ("Prod OpenAI", "db")
        assert (listed["from-config"]["display_name"], listed["from-config"]["source"]) == (None, "config")
        assert (by_name["display_name"], by_name["source"]) == ("Prod OpenAI", "db")
        assert config_by_name["source"] == "config"
        assert listed["openai-prod"]["credential_values"]["api_key"] != "sk-old"