From f1960ef5273f2701a9e221f7f3dd064dc25ab2c3 Mon Sep 17 00:00:00 2001 From: yucheng Date: Tue, 21 Jul 2026 07:05:03 +0000 Subject: [PATCH] test(proxy): cover credential database failures Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../credential_endpoints/test_endpoints.py | 66 +++++++++++++++++++ 1 file changed, 66 insertions(+) diff --git a/tests/test_litellm/proxy/credential_endpoints/test_endpoints.py b/tests/test_litellm/proxy/credential_endpoints/test_endpoints.py index 4b3f6e6f002..0563a2e9b5d 100644 --- a/tests/test_litellm/proxy/credential_endpoints/test_endpoints.py +++ b/tests/test_litellm/proxy/credential_endpoints/test_endpoints.py @@ -454,6 +454,72 @@ async def test_update_credential_handles_database_lookup_failure(_connected_db): _connected_db.update_by_name.assert_not_awaited() +@pytest.mark.asyncio +async def test_update_credential_handles_missing_database_client(monkeypatch): + import litellm.proxy.proxy_server as proxy_server + + monkeypatch.setattr(proxy_server, "prisma_client", None) + + result = await endpoints.update_credential( + request=MagicMock(), + fastapi_response=MagicMock(), + credential=UpdateCredentialItem( + credential_info={"access": {"global": True}}, + ), + credential_name="dest", + user_api_key_dict=_admin(), + ) + + assert result.code == "500" + + +@pytest.mark.asyncio +async def test_update_credential_handles_missing_database_credential(_connected_db): + _connected_db.find_by_name = AsyncMock(return_value=None) + _connected_db.update_by_name = AsyncMock() + + result = await endpoints.update_credential( + request=MagicMock(), + fastapi_response=MagicMock(), + credential=UpdateCredentialItem( + credential_info={"access": {"global": True}}, + ), + credential_name="dest", + user_api_key_dict=_admin(), + ) + + assert result.code == "404" + _connected_db.find_by_name.assert_awaited_once_with("dest") + _connected_db.update_by_name.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_update_credential_handles_database_write_failure(_connected_db): + db_error = ClientNotConnectedError() + existing = CredentialItem( + credential_name="dest", + credential_values={"langfuse_host": "h"}, + credential_info=_LOGGING_INFO, + ) + _connected_db.find_by_name = AsyncMock(return_value=existing) + _connected_db.update_by_name = AsyncMock(side_effect=db_error) + + result = await endpoints.update_credential( + request=MagicMock(), + fastapi_response=MagicMock(), + credential=UpdateCredentialItem( + credential_info={"access": {"global": True}}, + ), + credential_name="dest", + user_api_key_dict=_admin(), + ) + + assert result.code == "500" + assert result.message == str(db_error) + _connected_db.find_by_name.assert_awaited_once_with("dest") + _connected_db.update_by_name.assert_awaited_once() + + @pytest.mark.asyncio async def test_provider_credential_patch_forbidden_for_non_admin(_connected_db, monkeypatch): """A team-admin (or any non-admin) cannot PATCH a non-logging credential.