scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-05 02:41:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 49

test(proxy): cover narrow prisma connection error detection

Browse source
This commit is contained in:
Ishaan Jaffer 2026-02-20 17:44:18 -08:00
parent 4844f763cd
commit 3bb2423ab8
1 changed files with 20 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
tests/test_litellm/proxy/db/test_exception_handler.py
View file

@ -31,10 +31,28 @@ from litellm.proxy.db.exception_handler import PrismaDBExceptionHandler
# Test is_database_connection_error method
@pytest.mark.parametrize(
"prisma_error",
[
HTTPClientClosedError(),
ClientNotConnectedError(),
PrismaError("can't reach database server"),
PrismaError("connection refused"),
PrismaError("timed out while connecting"),
],
)
def test_is_database_connection_error_prisma_connection_errors(prisma_error):
"""
Test that only Prisma connection-related errors are considered DB connection errors.
"""
assert PrismaDBExceptionHandler.is_database_connection_error(prisma_error) == True
@pytest.mark.parametrize(
"prisma_error",
[
PrismaError(),
PrismaError("validation failed on query"),
DataError(data={"user_facing_error": {"meta": {"table": "test_table"}}}),
UniqueViolationError(
data={"user_facing_error": {"meta": {"table": "test_table"}}}
@ -52,15 +70,10 @@ from litellm.proxy.db.exception_handler import PrismaDBExceptionHandler
RecordNotFoundError(
data={"user_facing_error": {"meta": {"table": "test_table"}}}
),
HTTPClientClosedError(),
ClientNotConnectedError(),
],
)
def test_is_database_connection_error_prisma_errors(prisma_error):
"""
Test that all Prisma errors are considered database connection errors
"""
assert PrismaDBExceptionHandler.is_database_connection_error(prisma_error) == True
def test_is_database_connection_error_non_connection_prisma_errors(prisma_error):
assert PrismaDBExceptionHandler.is_database_connection_error(prisma_error) == False
def test_is_database_connection_generic_errors():