test(lint): ban blind pytest.raises(Exception) with ruff B017

ruff-tests.toml

@@ -14,11 +14,14 @@
 # B018    a bare attribute access or literal, usually a call missing its parens
 # PLW0127 `x = x` self-assignment, dead code that reads like a narrowing or a fixup
 # PLR0133 comparison of two constants, e.g. `assert True == True`
+# B017    `pytest.raises(Exception)` accepts the TypeError a refactor introduced just as
+#         readily as the rejection under test, so a crash reads as a pass. Narrow to the
+#         real type, or add `match=` where the code genuinely raises a bare Exception
 #
 # No target-version here on purpose: it resolves from requires-python (>=3.10), so
 # 3.11-only builtins like BaseExceptionGroup are correctly flagged in a tree that
 # still has to run on 3.10.
 
 line-length = 120
 
-lint.select = ["F821", "B011", "B015", "B018", "PT015", "PLR0133", "PLW0127"]
+lint.select = ["F821", "B011", "B015", "B017", "B018", "PT015", "PLR0133", "PLW0127"]

tests/guardrails_tests/test_bedrock_guardrails.py

@@ -197,14 +197,15 @@ async def test_bedrock_guardrails_block_responses_api():
 
 @pytest.mark.asyncio
 async def test_bedrock_guardrails_with_streaming():
+    from fastapi import HTTPException
     from litellm.proxy.utils import ProxyLogging
     from litellm.types.guardrails import GuardrailEventHooks
 
     # Create proper mock objects
     mock_user_api_key_cache = MagicMock(spec=DualCache)
     mock_user_api_key_dict = UserAPIKeyAuth()
 
-    with pytest.raises(Exception):  # Assert that this raises an exception
+    with pytest.raises(HTTPException):
         proxy_logging_obj = ProxyLogging(
             user_api_key_cache=mock_user_api_key_cache,
             premium_user=True,

tests/guardrails_tests/test_eu_ai_act_article5.py

@@ -20,6 +20,7 @@
 from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
     ContentFilterCategoryConfig,
 )
+from fastapi import HTTPException
 
 
 # Test cases: (sentence, expected_result, reason)
@@ -275,7 +276,7 @@ async def test_case_insensitive_matching(self, content_filter_guardrail):
         for sentence in sentences:
             request_data = {"messages": [{"role": "user", "content": sentence}]}
 
-            with pytest.raises(Exception):
+            with pytest.raises(HTTPException):
                 await content_filter_guardrail.apply_guardrail(
                     inputs={"texts": [sentence]},
                     request_data=request_data,
@@ -289,7 +290,7 @@ async def test_multiple_violations_in_one_sentence(self, content_filter_guardrai
         request_data = {"messages": [{"role": "user", "content": sentence}]}
 
         # Should block (contains multiple violations)
-        with pytest.raises(Exception):
+        with pytest.raises(HTTPException):
             await content_filter_guardrail.apply_guardrail(
                 inputs={"texts": [sentence]},
                 request_data=request_data,

tests/guardrails_tests/test_eu_ai_act_french_3_scenarios.py

@@ -19,6 +19,7 @@
 from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
     ContentFilterCategoryConfig,
 )
+from fastapi import HTTPException
 
 
 @pytest.fixture
@@ -228,7 +229,7 @@ async def test_mixed_french_english(self, content_filter_guardrail):
         request_data = {"messages": [{"role": "user", "content": sentence}]}
 
         # Should block (contains "build" and "système de crédit social")
-        with pytest.raises(Exception):
+        with pytest.raises(HTTPException):
             await content_filter_guardrail.apply_guardrail(
                 inputs={"texts": [sentence]},
                 request_data=request_data,
@@ -257,7 +258,7 @@ async def test_french_case_insensitive(self, content_filter_guardrail):
         request_data = {"messages": [{"role": "user", "content": sentence}]}
 
         # Should block (case-insensitive)
-        with pytest.raises(Exception):
+        with pytest.raises(HTTPException):
             await content_filter_guardrail.apply_guardrail(
                 inputs={"texts": [sentence]},
                 request_data=request_data,

tests/guardrails_tests/test_semantic_guard.py

@@ -10,6 +10,7 @@
 from unittest.mock import MagicMock
 
 import pytest
+from fastapi import HTTPException
 
 
 class TestRouteLoader:
@@ -307,7 +308,7 @@ def sql_injection_guardrail(self):
     @pytest.mark.asyncio
     async def test_sql_always_block(self, sql_injection_guardrail, sentence, reason):
         request_data = {"messages": [{"role": "user", "content": sentence}]}
-        with pytest.raises(Exception):
+        with pytest.raises(HTTPException):
             await sql_injection_guardrail.apply_guardrail(
                 inputs={"texts": [sentence]},
                 request_data=request_data,
@@ -343,7 +344,7 @@ async def test_sql_conditional_block(
         self, sql_injection_guardrail, sentence, reason
     ):
         request_data = {"messages": [{"role": "user", "content": sentence}]}
-        with pytest.raises(Exception):
+        with pytest.raises(HTTPException):
             await sql_injection_guardrail.apply_guardrail(
                 inputs={"texts": [sentence]},
                 request_data=request_data,
@@ -552,7 +553,7 @@ def content_filter_guardrail(self):
     @pytest.mark.asyncio
     async def test_always_block(self, content_filter_guardrail, sentence, reason):
         request_data = {"messages": [{"role": "user", "content": sentence}]}
-        with pytest.raises(Exception):
+        with pytest.raises(HTTPException):
             await content_filter_guardrail.apply_guardrail(
                 inputs={"texts": [sentence]},
                 request_data=request_data,

tests/image_gen_tests/test_image_generation.py

@@ -458,7 +458,7 @@ async def test_azure_image_generation_request_body():
     ) as mock_post:
         mock_post.side_effect = Exception("test")
 
-        with pytest.raises(Exception):
+        with pytest.raises(litellm.APIConnectionError):
             await aimage_generation(
                 model="azure/gpt-image-1",
                 prompt="test prompt",

tests/litellm_utils_tests/test_utils.py

@@ -1334,7 +1334,7 @@ def test_validate_chat_completion_user_messages(messages, expected_bool):
         validate_chat_completion_user_messages(messages=messages)
     else:
         ## Invalid message
-        with pytest.raises(Exception):
+        with pytest.raises(Exception, match="Invalid user message at index 0"):
             validate_chat_completion_user_messages(messages=messages)
 
 
@@ -1354,7 +1354,7 @@ def test_validate_chat_completion_tool_choice(tool_choice, expected_bool):
     if expected_bool:
         validate_chat_completion_tool_choice(tool_choice=tool_choice)
     else:
-        with pytest.raises(Exception):
+        with pytest.raises(Exception, match="Invalid tool choice"):
             validate_chat_completion_tool_choice(tool_choice=tool_choice)
 
 

tests/llm_responses_api_testing/base_responses_api.py

@@ -28,6 +28,7 @@
     ResponseInputParam,
 )
 from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
+import openai
 
 
 def validate_responses_api_response(response, final_chunk: bool = False):
@@ -700,12 +701,12 @@ async def test_cancel_responses_invalid_response_id(self, sync_mode):
         base_completion_call_args = self.get_base_completion_call_args()
 
         if sync_mode:
-            with pytest.raises(Exception):
+            with pytest.raises(openai.APIError):
                 litellm.cancel_responses(
                     response_id="invalid_response_id_12345", **base_completion_call_args
                 )
         else:
-            with pytest.raises(Exception):
+            with pytest.raises(openai.APIError):
                 await litellm.acancel_responses(
                     response_id="invalid_response_id_12345", **base_completion_call_args
                 )

tests/llm_translation/test_llm_response_utils/test_convert_dict_to_chat_completion.py

@@ -1243,7 +1243,7 @@ def test_convert_to_model_response_object_with_error_code_only():
         },
     }
 
-    with pytest.raises(Exception):
+    with pytest.raises(Exception) as exc_info:  # noqa: B017  # bare Exception raised, so status_code is the assertion
         convert_to_model_response_object(
             model_response_object=ModelResponse(),
             response_object=response_object,
@@ -1255,6 +1255,8 @@ def test_convert_to_model_response_object_with_error_code_only():
             convert_tool_call_to_json_mode=False,
         )
 
+    assert exc_info.value.status_code == 500
+
 
 def test_model_prefix_preservation():
     """
@@ -2473,14 +2475,14 @@ def test_reasoning_content_not_mirrored_into_provider_specific_fields(self):
         assert "reasoning_content" not in (message.provider_specific_fields or {})
 
     def test_response_none_raises(self):
-        with pytest.raises(Exception):
+        with pytest.raises(Exception, match="Invalid response object"):
             convert_to_model_response_object(
                 response_object=None,
                 model_response_object=ModelResponse(),
             )
 
     def test_model_response_none_raises(self):
-        with pytest.raises(Exception):
+        with pytest.raises(Exception, match="Invalid response object"):
             convert_to_model_response_object(
                 response_object={
                     "choices": [

tests/llm_translation/test_openai.py

@@ -1458,7 +1458,7 @@ def test_responses_gpt54_with_xhigh_reasoning():
         # Stop execution right after request generation to avoid external API calls.
         mock_responses.side_effect = RuntimeError("stop_after_request_build")
 
-        with pytest.raises(Exception):
+        with pytest.raises(litellm.APIConnectionError):
             litellm.completion(
                 model="openai/responses/gpt-5.4",
                 messages=[{"role": "user", "content": "What is 2+2?"}],

tests/local_testing/test_get_llm_provider.py

@@ -569,5 +569,5 @@ def test_routing_comes_from_the_rule_not_python(self, shipped_generalizations):
         )
 
         set_fallback_generalizations([])
-        with pytest.raises(Exception):
+        with pytest.raises(litellm.BadRequestError):
             litellm.get_llm_provider(model="claude-opus-4-9")

tests/openai_endpoints_tests/test_e2e_openai_responses_api.py

@@ -1,5 +1,5 @@
 import httpx
-from openai import OpenAI, BadRequestError
+from openai import OpenAI, BadRequestError, APIStatusError
 import pytest
 
 
@@ -87,7 +87,7 @@ def test_basic_response():
     print("DELETE response=", delete_response)
 
     # expect an error when getting the response again since it was deleted
-    with pytest.raises(Exception):
+    with pytest.raises(APIStatusError):
         get_response = client.responses.retrieve(response.id)
 
 
@@ -195,6 +195,6 @@ def test_cancel_streaming_response():
 
 def test_cancel_invalid_response_id():
     client = get_test_client()
-    with pytest.raises(Exception):
+    with pytest.raises(APIStatusError):
         # Try to cancel a non-existent response ID
         client.responses.cancel("invalid_response_id_12345")

tests/proxy_unit_tests/test_auth_checks.py

@@ -330,6 +330,7 @@ async def test_wildcard_access_after_cost_map_reload(key_models, model, expect_t
     Fix: each reload now calls litellm.add_known_models(model_cost_map=new_map)
     with the fetched map passed explicitly to avoid any reference ambiguity.
     """
+    from litellm.proxy._types import ProxyException
     from litellm.proxy.auth.auth_checks import can_key_call_model
 
     # Build a new cost map that includes the brand-new model — exactly what
@@ -378,7 +379,7 @@ async def test_wildcard_access_after_cost_map_reload(key_models, model, expect_t
                 llm_router=router,
             )
         else:
-            with pytest.raises(Exception):
+            with pytest.raises(ProxyException):
                 await can_key_call_model(
                     model=model,
                     llm_model_list=llm_model_list,

tests/proxy_unit_tests/test_jwt.py

@@ -41,6 +41,7 @@
 from litellm.proxy.management_endpoints.team_endpoints import new_team
 from litellm.proxy.proxy_server import chat_completion
 from typing import Literal, Optional
+from litellm.proxy._types import ProxyException
 
 public_key = {
     "kty": "RSA",
@@ -1045,9 +1046,7 @@ async def test_allow_access_by_email(
             assert result is not None  # Adjust this based on your actual response check
         else:
             # Expect the call to fail
-            with pytest.raises(
-                Exception
-            ):  # Replace with the actual exception raised on failure
+            with pytest.raises(ProxyException):
                 resp = await user_api_key_auth(request=request, api_key=bearer_token)
                 print(resp)
 

tests/proxy_unit_tests/test_proxy_config_unit_test.py

@@ -53,7 +53,7 @@ async def test_read_config_from_bad_file_path():
     """
     proxy_config_instance = ProxyConfig()
     config_path = "non-existent-file.yaml"
-    with pytest.raises(Exception):
+    with pytest.raises(Exception, match="Config file not found"):
         config = await proxy_config_instance.get_config(config_file_path=config_path)
 
 

tests/proxy_unit_tests/test_proxy_utils.py

@@ -29,6 +29,7 @@
     _get_dynamic_logging_metadata,
     add_litellm_data_to_request,
 )
+from pydantic import ValidationError
 
 pytestmark = pytest.mark.xdist_group("proxy_heavy")
 
@@ -1695,13 +1696,13 @@ def test_update_key_request_validation():
     """
     from litellm.proxy._types import UpdateKeyRequest
 
-    with pytest.raises(Exception):
+    with pytest.raises(ValidationError):
         UpdateKeyRequest(
             key="test_key",
             temp_budget_increase=100,
         )
 
-    with pytest.raises(Exception):
+    with pytest.raises(ValidationError):
         UpdateKeyRequest(
             key="test_key",
             temp_budget_expiry="2024-01-20T00:00:00Z",
@@ -1848,7 +1849,7 @@ async def test_end_user_transactions_reset():
     mock_client.db.tx = AsyncMock(side_effect=Exception("DB Error"))
 
     # Call function - should raise error
-    with pytest.raises(Exception):
+    with pytest.raises(TypeError):
         await ProxyUpdateSpend.update_end_user_spend(
             n_retry_times=0,
             prisma_client=mock_client,
@@ -1878,7 +1879,7 @@ async def test_spend_logs_cleanup_after_error():
     original_logs = mock_client.spend_log_transactions.copy()
 
     # Call function - should raise error
-    with pytest.raises(Exception):
+    with pytest.raises(TypeError):
         await ProxyUpdateSpend.update_spend_logs(
             n_retry_times=0,
             prisma_client=mock_client,

tests/proxy_unit_tests/test_skills_db.py

@@ -26,6 +26,7 @@
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.utils import PrismaClient, ProxyLogging
 from litellm.types.utils import LlmProviders
+import openai
 
 proxy_logging_obj = ProxyLogging(user_api_key_cache=DualCache())
 
@@ -254,7 +255,7 @@ async def test_delete_skill_sdk(prisma_client):
     assert result.type == "skill_deleted"
 
     # Verify skill no longer exists
-    with pytest.raises(Exception):
+    with pytest.raises(openai.APIError):
         await aget_skill(
             skill_id=created_skill.id,
             custom_llm_provider=LlmProviders.LITELLM_PROXY.value,

tests/proxy_unit_tests/test_user_api_key_auth.py

@@ -436,7 +436,7 @@ def test_ui_token_route_access(route, user_role, should_be_allowed):
         )
         assert result is True
     else:
-        with pytest.raises(Exception):
+        with pytest.raises(Exception, match="Only proxy admin can be used to generate"):
             _is_api_route_allowed(
                 route=route,
                 request=request,

tests/test_litellm/caching/test_redis_cache.py

@@ -520,13 +520,15 @@ async def test_circuit_breaker_covers_lua_script_execution(redis_no_ping):
     counted toward taking Redis out of the pool and kept paying a full socket timeout
     each, which is the traffic the outage hurts most.
     """
+    from redis.exceptions import ConnectionError as RedisConnectionError
+
     from litellm.constants import REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD
 
     cache = RedisCache(host="127.0.0.1", port=_closed_port(), socket_timeout=0.5)
     run_script = cache.async_register_script("return 1")
 
     for _ in range(REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD):
-        with pytest.raises(Exception):
+        with pytest.raises(RedisConnectionError):
             await run_script(keys=["lit4930"], args=[1])
 
     with pytest.raises(Exception, match="circuit breaker is open"):
@@ -603,8 +605,11 @@ async def test_only_connectivity_failures_open_the_breaker(error, opens_breaker)
     async def failing_call():
         raise raised
 
-    for _ in range(breaker.failure_threshold + 1):
-        with pytest.raises(Exception):
+    for _ in range(breaker.failure_threshold):
+        with pytest.raises(type(raised)):
             await _run_under_circuit_breaker(breaker, "op", failing_call)
 
+    with pytest.raises(Exception, match="circuit breaker is open" if opens_breaker else "boom"):
+        await _run_under_circuit_breaker(breaker, "op", failing_call)
+
     assert breaker.is_open() is opens_breaker

tests/test_litellm/containers/test_container_api.py

@@ -384,7 +384,7 @@ def test_create_container_error_handling(self):
             "container_create_handler",
             side_effect=Exception("API Error"),
         ):
-            with pytest.raises(Exception):
+            with pytest.raises(litellm.APIConnectionError):
                 create_container(
                     name="Error Test Container", custom_llm_provider="openai"
                 )