From f2883192cd0b13c1b9b91e835e4b5191ba3573ef Mon Sep 17 00:00:00 2001 From: Ishaan Jaffer Date: Fri, 20 Feb 2026 17:44:23 -0800 Subject: [PATCH] test(proxy): add reconnect lock-timeout behavior coverage --- .../proxy/db/test_prisma_self_heal.py | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/tests/test_litellm/proxy/db/test_prisma_self_heal.py b/tests/test_litellm/proxy/db/test_prisma_self_heal.py index 3abecf49751..762493f6156 100644 --- a/tests/test_litellm/proxy/db/test_prisma_self_heal.py +++ b/tests/test_litellm/proxy/db/test_prisma_self_heal.py @@ -66,6 +66,32 @@ async def test_attempt_db_reconnect_should_skip_when_in_cooldown(mock_proxy_logg client.db.query_raw.assert_not_called() +@pytest.mark.asyncio +async def test_attempt_db_reconnect_should_skip_when_lock_timeout_expires( + mock_proxy_logging, +): + client = PrismaClient(database_url="mock://test", proxy_logging_obj=mock_proxy_logging) + client.db.disconnect = AsyncMock(return_value=None) + client.db.connect = AsyncMock(return_value=None) + client.db.query_raw = AsyncMock(return_value=[{"result": 1}]) + + await client._db_reconnect_lock.acquire() + try: + result = await client.attempt_db_reconnect( + reason="unit_test_reconnect_lock_timeout", + force=True, + timeout_seconds=0.1, + lock_timeout_seconds=0.01, + ) + finally: + client._db_reconnect_lock.release() + + assert result is False + client.db.disconnect.assert_not_called() + client.db.connect.assert_not_called() + client.db.query_raw.assert_not_called() + + @pytest.mark.asyncio async def test_attempt_db_reconnect_should_set_cooldown_after_attempt(mock_proxy_logging): client = PrismaClient(database_url="mock://test", proxy_logging_obj=mock_proxy_logging)