Merge 8d821edfa5 into 86ef1fb08b

scripts/check_test_quality.py

@@ -52,6 +52,17 @@ TQ008   A `patch(...)` whose target is a `litellm.` internal. Patching the SDK's
         functions pins the test to the current wiring instead of the behaviour, and it
         is the idiom the suite reaches for instead of faking the HTTP boundary. Mocking
         a third-party client, a transport, or anything outside `litellm.` is untouched.
+TQ009   An `assert` inside a `try` whose handler catches `AssertionError`. A failing
+        assertion raises `AssertionError`, which is an `Exception`, so a bare `except:`
+        or an `except Exception` in the same test catches it. The handler returns
+        normally, and pytest records a pass on exactly the regression the assertion was
+        written to catch. A handler that re-raises, or that calls `pytest.fail`, is
+        reporting the failure and is untouched; `pytest.skip` is not, since it turns a
+        real regression into a skip that reads as somebody's deliberate guard. Put the
+        assertions below the block, or narrow the handler to the error the call under
+        test actually raises. `pytest.raises` and `pytest.fail` inside the body are
+        untouched too: `Failed` inherits from `BaseException`, so `except Exception`
+        never sees them.
 TQ007   A module global that a conftest saves before every test and restores after it.
         The save/restore list is a hand-maintained inventory of the leaks the suite
         already knows about, so it is allowed to shrink and never to grow: a new entry
@@ -112,6 +123,7 @@ import sys
 import tokenize
 from collections.abc import Iterable, Iterator, Mapping, Sequence
 from dataclasses import dataclass
+from itertools import takewhile
 from multiprocessing import Pool
 from pathlib import Path
 from types import MappingProxyType
@@ -134,6 +146,8 @@ MOCK_ASSERTION_PREFIX: Final = "assert_"
 
 PATCH_MEMBERS: Final = frozenset(("object", "dict", "multiple"))
 
+ASSERTION_ERROR_CATCHERS: Final = frozenset(("Exception", "BaseException", "AssertionError"))
+
 ENVIRON_READERS: Final = frozenset(("os.environ.get", "environ.get", "os.getenv", "getenv"))
 ENVIRON_MAPPINGS: Final = frozenset(("os.environ", "environ"))
 SKIP_CALLS: Final = frozenset(("pytest.skip", "skip"))
@@ -406,6 +420,108 @@ def iter_assertion_violations(path: Path, tree: ast.Module) -> Iterator[Violatio
             )
 
 
+def _raises_assertion_error(nodes: Iterable[ast.AST]) -> bool:
+    """An `assert` or an `assert*()` call, the two shapes that fail via AssertionError.
+    `pytest.raises`/`fail` are excluded: `Failed` inherits from BaseException, so an
+    `except Exception` never catches them."""
+    return any(
+        isinstance(node, ast.Assert)
+        or (isinstance(node, ast.Call) and _is_assertion_helper_call(node))
+        for parent in nodes
+        for node in ast.walk(parent)
+    )
+
+
+def _catches_assertion_error(handler: ast.ExceptHandler) -> bool:
+    if handler.type is None:
+        return True
+    named: Final = handler.type.elts if isinstance(handler.type, ast.Tuple) else (handler.type,)
+    return any(_dotted_name(node).rpartition(".")[2] in ASSERTION_ERROR_CATCHERS for node in named)
+
+
+def _walk_within_scope(node: ast.AST) -> Iterator[ast.AST]:
+    """`ast.walk` that stops at a nested function or class, whose body is a scope of its
+    own rather than more of the statements around it."""
+    if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)):
+        return
+    yield node
+    for child in ast.iter_child_nodes(node):
+        yield from _walk_within_scope(child)
+
+
+def _statement_reports(stmt: ast.stmt) -> bool:
+    """Whether control leaving this statement has necessarily reported the failure.
+    A `raise` or a `pytest.fail` does. An `if` does only when both halves do, since the
+    branch that falls through is the path a swallowed assertion takes."""
+    if isinstance(stmt, (ast.Raise, ast.Assert)):
+        return True
+    if isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Call):
+        return _is_pytest_assertion_call(stmt.value) or _is_assertion_helper_call(stmt.value)
+    if isinstance(stmt, ast.If):
+        return bool(stmt.orelse) and _reports_the_failure(stmt.body) and _reports_the_failure(stmt.orelse)
+    if isinstance(stmt, (ast.With, ast.AsyncWith)):
+        return _reports_the_failure(stmt.body)
+    if isinstance(stmt, ast.Try):
+        return _reports_the_failure(stmt.body) or _reports_the_failure(stmt.finalbody)
+    if isinstance(stmt, ast.Match):
+        return bool(stmt.cases) and all(_reports_the_failure(case.body) for case in stmt.cases)
+    return False
+
+
+def _statement_escapes(stmt: ast.stmt) -> bool:
+    """Whether control can leave the enclosing handler through this statement without
+    having reported. A `return`, `break` or `continue` on some branch does exactly that,
+    and it makes every reporting statement after it unreachable on that path. A nested
+    `def` is not walked into, since its `return` leaves the nested body, not the handler."""
+    return not _statement_reports(stmt) and any(
+        isinstance(node, (ast.Return, ast.Break, ast.Continue))
+        for node in _walk_within_scope(stmt)
+    )
+
+
+def _reports_the_failure(body: Sequence[ast.stmt]) -> bool:
+    """Re-raising, or failing the test, passes the failure on rather than eating it.
+    Every path out of the block has to do it, so the scan stops at the first statement
+    that can escape without reporting: a `raise` reachable on one branch only, or sitting
+    below an early `return`, leaves the other path swallowing, which is the shape the
+    rule exists to catch."""
+    return any(
+        _statement_reports(stmt)
+        for stmt in takewhile(lambda candidate: not _statement_escapes(candidate), body)
+    )
+
+
+def _swallowing_handler(node: ast.Try) -> ast.ExceptHandler | None:
+    if not _raises_assertion_error(node.body):
+        return None
+    return next(
+        (
+            handler
+            for handler in node.handlers
+            if _catches_assertion_error(handler) and not _reports_the_failure(handler.body)
+        ),
+        None,
+    )
+
+
+def iter_swallowed_assertion_violations(path: Path, tree: ast.Module) -> Iterator[Violation]:
+    for function in iter_test_functions(tree):
+        for node in ast.walk(function):
+            if not isinstance(node, ast.Try):
+                continue
+            handler: Final = _swallowing_handler(node)
+            if handler is not None:
+                yield Violation(
+                    path,
+                    handler.lineno,
+                    "TQ009",
+                    f"an assertion in this `try` fails with AssertionError, which `{function.name}` "
+                    "catches here and discards, so the test reports green on the regression it was "
+                    "written to catch; move the assertions below the block or narrow the handler "
+                    f"(suppress: `# {SUPPRESSION_TOKEN}: <reason>`)",
+                )
+
+
 def iter_sys_path_violations(path: Path, tree: ast.Module) -> Iterator[Violation]:
     for node in ast.walk(tree):
         if isinstance(node, ast.Call) and _dotted_name(node.func) == "sys.path.insert":
@@ -740,6 +856,7 @@ def check_file(path: Path) -> tuple[Violation, ...]:
         violation
         for violation in (
             *iter_assertion_violations(path, tree),
+            *iter_swallowed_assertion_violations(path, tree),
             *iter_sys_path_violations(path, tree),
             *iter_environ_violations(path, tree),
             *iter_global_mutation_violations(path, tree),

test-quality-budget.json

@@ -22,5 +22,8 @@
   },
   "TQ008": {
     "limit": 11139
+  },
+  "TQ009": {
+    "limit": 44
   }
 }

tests/test_litellm/litellm_core_utils/test_litellm_logging.py

@@ -61,31 +61,15 @@ def test_post_call_serializes_dict_with_datetime(logging_obj):
 
 
 def test_sentry_sample_rate(monkeypatch):
-    existing_sample_rate = os.getenv("SENTRY_API_SAMPLE_RATE")
-    try:
-        # test with default value by removing the environment variable
-        if existing_sample_rate:
-            del os.environ["SENTRY_API_SAMPLE_RATE"]
+    import sentry_sdk
 
-        set_callbacks(["sentry"])
-        # Check if the default sample rate is set to 1.0
-        assert os.environ.get("SENTRY_API_SAMPLE_RATE") == "1.0"
+    monkeypatch.delenv("SENTRY_API_SAMPLE_RATE", raising=False)
+    set_callbacks(["sentry"])
+    assert sentry_sdk.get_client().options["sample_rate"] == 1.0
 
-        # test with custom value
-        monkeypatch.setenv("SENTRY_API_SAMPLE_RATE", "0.5")
-
-        set_callbacks(["sentry"])
-        # Check if the custom sample rate is set correctly
-        assert os.environ.get("SENTRY_API_SAMPLE_RATE") == "0.5"
-    except Exception as e:
-        print(f"Error: {e}")
-    finally:
-        # Restore the original environment variable
-        if existing_sample_rate:
-            monkeypatch.setenv("SENTRY_API_SAMPLE_RATE", existing_sample_rate)
-        else:
-            if "SENTRY_API_SAMPLE_RATE" in os.environ:
-                del os.environ["SENTRY_API_SAMPLE_RATE"]
+    monkeypatch.setenv("SENTRY_API_SAMPLE_RATE", "0.5")
+    set_callbacks(["sentry"])
+    assert sentry_sdk.get_client().options["sample_rate"] == 0.5
 
 
 def test_sentry_environment(monkeypatch):
@@ -106,8 +90,8 @@ def test_sentry_environment(monkeypatch):
     mock_sentry_sdk.init = mock_init
 
     # Inject mocks into sys.modules
-    sys.modules["sentry_sdk"] = mock_sentry_sdk
-    sys.modules["sentry_sdk.scrubber"] = mock_scrubber_module
+    monkeypatch.setitem(sys.modules, "sentry_sdk", mock_sentry_sdk)
+    monkeypatch.setitem(sys.modules, "sentry_sdk.scrubber", mock_scrubber_module)
 
     try:
         # Set a mock DSN to allow Sentry initialization
@@ -2147,8 +2131,8 @@ def test_sentry_event_scrubber_initialization(monkeypatch):
     mock_sentry_sdk.init = mock_init
 
     # Step 3: Inject both into sys.modules BEFORE import occurs
-    sys.modules["sentry_sdk"] = mock_sentry_sdk
-    sys.modules["sentry_sdk.scrubber"] = mock_scrubber_module
+    monkeypatch.setitem(sys.modules, "sentry_sdk", mock_sentry_sdk)
+    monkeypatch.setitem(sys.modules, "sentry_sdk.scrubber", mock_scrubber_module)
 
     # Step 4: Run the actual sentry setup code
     set_callbacks(["sentry"])

tests/test_litellm/llms/vertex_ai/gemini/test_vertex_ai_gemini_transformation.py

@@ -1669,12 +1669,29 @@ def test_gemini_history_nests_multimodal_tool_response_parts():
     ]
 
 
-def test_convert_tool_response_with_url_image():
+def test_convert_tool_response_with_url_image(monkeypatch):
     """Test tool response with HTTP URL image (will download and convert)."""
-    import pytest
+    import httpx
 
-    # Use a publicly accessible test image URL
-    test_image_url = "https://via.placeholder.com/1x1.png"
+    import litellm
+
+    png_bytes = base64.b64decode(
+        "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
+    )
+    monkeypatch.setattr(
+        litellm.module_level_client,
+        "client",
+        httpx.Client(
+            transport=httpx.MockTransport(
+                lambda request: httpx.Response(
+                    200, content=png_bytes, headers={"Content-Type": "image/png"}
+                )
+            )
+        ),
+    )
+
+    # A literal public IP keeps the SSRF check off DNS; MockTransport answers the request
+    test_image_url = "https://1.1.1.1/screenshot.png"
 
     tool_message = {
         "role": "tool",
@@ -1697,30 +1714,24 @@ def test_convert_tool_response_with_url_image():
         ]
     }
 
-    try:
-        result = convert_to_gemini_tool_call_result(
-            tool_message, last_message_with_tool_calls
-        )
+    result = convert_to_gemini_tool_call_result(
+        tool_message, last_message_with_tool_calls
+    )
 
-        assert isinstance(
-            result, list
-        ), "Should return a parts list when media is present"
-        assert len(result) == 1, "Should return one function_response part"
-        result_part = result[0]
-        assert "function_response" in result_part
-        assert "inline_data" not in result_part
-        function_response = result_part["function_response"]
-        assert function_response["name"] == "type_text_at"
+    assert isinstance(result, list), "Should return a parts list when media is present"
+    assert len(result) == 1, "Should return one function_response part"
+    result_part = result[0]
+    assert "function_response" in result_part
+    assert "inline_data" not in result_part
+    function_response = result_part["function_response"]
+    assert function_response["name"] == "type_text_at"
 
-        # Check inline_data is nested under functionResponse.parts.
-        assert "parts" in function_response
-        assert len(function_response["parts"]) == 1
-        inline_data: BlobType = function_response["parts"][0]["inline_data"]
-        assert "data" in inline_data
-        assert "mime_type" in inline_data
-    except Exception as e:
-        # Skip test if URL download fails (no internet connection, etc.)
-        pytest.skip(f"Failed to download image from URL: {e}")
+    # Check inline_data is nested under functionResponse.parts.
+    assert "parts" in function_response
+    assert len(function_response["parts"]) == 1
+    inline_data: BlobType = function_response["parts"][0]["inline_data"]
+    assert "data" in inline_data
+    assert "mime_type" in inline_data
 
 
 def test_convert_tool_response_text_only():

tests/test_litellm/proxy/pass_through_endpoints/test_pass_through_endpoints.py

@@ -2574,21 +2574,19 @@ async def test_pass_through_with_httpbin_redirect():
             custom_headers={},
             user_api_key_dict=mock_user_api_key_dict,
         )
-
-        # Should get the final response (200) from /get endpoint, not the redirect (302)
-        assert response.status_code == 200
-
-        # The response should be from the /get endpoint
-        response_content = bytes(response.body).decode("utf-8")
-
-        # httpbin.org/get returns JSON with info about the request
-        assert '"url": "https://httpbin.org/get"' in response_content
     except Exception as e:
         # If httpbin.org is not accessible, skip the test
-        import pytest
-
         pytest.skip(f"Could not reach httpbin.org for integration test: {e}")
 
+    # Should get the final response (200) from /get endpoint, not the redirect (302)
+    assert response.status_code == 200
+
+    # The response should be from the /get endpoint
+    response_content = bytes(response.body).decode("utf-8")
+
+    # httpbin.org/get returns JSON with info about the request
+    assert '"url": "https://httpbin.org/get"' in response_content
+
 
 @pytest.mark.asyncio
 async def test_filter_endpoints_by_team_allowed_routes_with_filter():

tests/test_litellm/test_check_test_quality.py

@@ -681,6 +681,329 @@ def test_an_sdk_patch_can_be_suppressed(tmp_path):
     assert "TQ008" not in _codes(tmp_path, source)
 
 
+def _swallowed(tmp_path, body):
+    return _codes(tmp_path, "import pytest\n\n\ndef test_x():\n" + body)
+
+
+def test_an_assert_caught_by_except_exception_is_flagged(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert compute() == 3\n"
+        "    except Exception:\n"
+        "        pass\n"
+    )
+    assert _swallowed(tmp_path, body) == ["TQ009"]
+
+
+def test_a_bare_except_swallows_it_just_the_same(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert compute() == 3\n"
+        "    except:\n"
+        "        print('oh well')\n"
+    )
+    assert _swallowed(tmp_path, body) == ["TQ009"]
+
+
+def test_naming_assertion_error_directly_is_flagged(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert compute() == 3\n"
+        "    except AssertionError:\n"
+        "        pass\n"
+    )
+    assert _swallowed(tmp_path, body) == ["TQ009"]
+
+
+def test_a_tuple_handler_is_flagged_when_any_member_catches_it(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert compute() == 3\n"
+        "    except (ValueError, Exception):\n"
+        "        pass\n"
+    )
+    assert _swallowed(tmp_path, body) == ["TQ009"]
+
+
+def test_a_skip_reports_the_failure_as_somebody_elses_guard(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert compute() == 3\n"
+        "    except Exception as e:\n"
+        "        pytest.skip(f'no network: {e}')\n"
+    )
+    assert _swallowed(tmp_path, body) == ["TQ009"]
+
+
+def test_an_assertion_helper_call_counts_as_the_assertion(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert_auth_denied(call(), 'missing header')\n"
+        "    except Exception:\n"
+        "        pass\n"
+    )
+    assert _swallowed(tmp_path, body) == ["TQ009"]
+
+
+def test_a_nested_assert_below_a_loop_is_still_inside_the_block(tmp_path):
+    body = (
+        "    try:\n"
+        "        for item in items:\n"
+        "            assert item.ok\n"
+        "    except Exception:\n"
+        "        pass\n"
+    )
+    assert _swallowed(tmp_path, body) == ["TQ009"]
+
+
+def test_a_narrow_handler_that_cannot_see_assertion_error_is_left_alone(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert compute() == 3\n"
+        "    except ValueError:\n"
+        "        pass\n"
+    )
+    assert _swallowed(tmp_path, body) == []
+
+
+def test_a_handler_that_reraises_is_left_alone(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert compute() == 3\n"
+        "    except Exception:\n"
+        "        print('context')\n"
+        "        raise\n"
+    )
+    assert _swallowed(tmp_path, body) == []
+
+
+def test_a_handler_that_fails_the_test_is_left_alone(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert compute() == 3\n"
+        "    except Exception as e:\n"
+        "        pytest.fail(f'boom: {e}')\n"
+    )
+    assert _swallowed(tmp_path, body) == []
+
+
+def test_a_handler_that_asserts_on_the_error_is_left_alone(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert compute() == 3\n"
+        "    except Exception as e:\n"
+        "        assert 'boom' in str(e)\n"
+    )
+    assert _swallowed(tmp_path, body) == []
+
+
+def test_a_try_block_with_no_assertion_in_it_is_left_alone(tmp_path):
+    body = (
+        "    try:\n"
+        "        value = compute()\n"
+        "    except Exception:\n"
+        "        pytest.skip('no network')\n"
+        "    assert value == 3\n"
+    )
+    assert _swallowed(tmp_path, body) == []
+
+
+def test_pytest_raises_in_the_body_escapes_an_except_exception(tmp_path):
+    body = (
+        "    try:\n"
+        "        with pytest.raises(ValueError):\n"
+        "            compute()\n"
+        "    except Exception:\n"
+        "        pass\n"
+    )
+    assert _swallowed(tmp_path, body) == []
+
+
+def test_pytest_fail_in_the_body_escapes_an_except_exception(tmp_path):
+    body = (
+        "    try:\n"
+        "        if compute() != 3:\n"
+        "            pytest.fail('wrong')\n"
+        "    except Exception:\n"
+        "        pass\n"
+    )
+    assert _swallowed(tmp_path, body) == []
+
+
+def test_the_swallow_is_reported_at_the_handler_not_the_try(tmp_path):
+    snippet = tmp_path / "test_snippet.py"
+    snippet.write_text(
+        "def test_x():\n"
+        "    try:\n"
+        "        assert compute() == 3\n"
+        "    except Exception:\n"
+        "        pass\n",
+        encoding="utf-8",
+    )
+    swallowed = [v for v in checker.check_file(snippet) if v.code == "TQ009"]
+    assert [v.line for v in swallowed] == [4]
+
+
+def test_only_the_swallowing_handler_of_several_is_reported(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert compute() == 3\n"
+        "    except ValueError:\n"
+        "        raise\n"
+        "    except Exception:\n"
+        "        pass\n"
+    )
+    assert _swallowed(tmp_path, body) == ["TQ009"]
+
+
+def test_the_swallow_is_suppressible_with_a_reason(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert compute() == 3\n"
+        "    except Exception:  # test-quality-ok: the flake is tracked in LIT-1234\n"
+        "        pass\n"
+    )
+    assert _swallowed(tmp_path, body) == []
+
+
+def test_a_raise_reachable_on_only_one_branch_still_swallows(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert f() == expected\n"
+        "    except Exception as e:\n"
+        "        if expected is False:\n"
+        "            pass\n"
+        "        else:\n"
+        "            raise e\n"
+    )
+    assert _swallowed(tmp_path, body) == ["TQ009"]
+
+
+def test_a_branch_that_reports_on_both_halves_is_left_alone(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert f() == 3\n"
+        "    except Exception as e:\n"
+        "        if strict:\n"
+        "            pytest.fail(str(e))\n"
+        "        else:\n"
+        "            raise\n"
+    )
+    assert _swallowed(tmp_path, body) == []
+
+
+def test_an_if_with_no_else_falls_through_and_swallows(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert f() == 3\n"
+        "    except Exception:\n"
+        "        if strict:\n"
+        "            raise\n"
+    )
+    assert _swallowed(tmp_path, body) == ["TQ009"]
+
+
+def test_an_elif_chain_that_ends_in_a_swallowing_branch_is_flagged(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert f() == 3\n"
+        "    except Exception as e:\n"
+        "        if 'timeout' in str(e):\n"
+        "            pass\n"
+        "        elif 'refused' in str(e):\n"
+        "            pass\n"
+        "        else:\n"
+        "            pytest.fail(str(e))\n"
+    )
+    assert _swallowed(tmp_path, body) == ["TQ009"]
+
+
+def test_a_report_nested_in_a_with_block_still_counts(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert f() == 3\n"
+        "    except Exception:\n"
+        "        with context():\n"
+        "            raise\n"
+    )
+    assert _swallowed(tmp_path, body) == []
+
+
+def test_a_report_after_a_swallowing_branch_still_counts(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert f() == 3\n"
+        "    except Exception as e:\n"
+        "        if noisy:\n"
+        "            print(e)\n"
+        "        raise\n"
+    )
+    assert _swallowed(tmp_path, body) == []
+
+
+def test_an_early_return_below_a_raise_still_swallows(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert f() == 3\n"
+        "    except Exception as e:\n"
+        "        if 'flaky' in str(e):\n"
+        "            return\n"
+        "        raise\n"
+    )
+    assert _swallowed(tmp_path, body) == ["TQ009"]
+
+
+def test_a_continue_below_a_pytest_fail_still_swallows(tmp_path):
+    body = (
+        "    for item in items:\n"
+        "        try:\n"
+        "            assert f(item) == 3\n"
+        "        except Exception as e:\n"
+        "            if 'skip' in str(e):\n"
+        "                continue\n"
+        "            pytest.fail(str(e))\n"
+    )
+    assert _swallowed(tmp_path, body) == ["TQ009"]
+
+
+def test_a_break_below_a_raise_still_swallows(tmp_path):
+    body = (
+        "    for item in items:\n"
+        "        try:\n"
+        "            assert f(item) == 3\n"
+        "        except Exception:\n"
+        "            if done:\n"
+        "                break\n"
+        "            raise\n"
+    )
+    assert _swallowed(tmp_path, body) == ["TQ009"]
+
+
+def test_a_return_in_a_nested_def_does_not_escape_the_handler(tmp_path):
+    body = (
+        "    try:\n"
+        "        assert f() == 3\n"
+        "    except Exception:\n"
+        "        def _later():\n"
+        "            return 1\n"
+        "        raise\n"
+    )
+    assert _swallowed(tmp_path, body) == []
+
+
+def test_a_swallow_outside_a_test_function_is_left_alone(tmp_path):
+    source = (
+        "def _helper():\n"
+        "    try:\n"
+        "        assert compute() == 3\n"
+        "    except Exception:\n"
+        "        pass\n\n\n"
+        "def test_x():\n"
+        "    assert _helper() is None\n"
+    )
+    assert _codes(tmp_path, source) == []
+
+
 _FANS_OUT = checker._worker_count(checker.PARALLEL_MIN_PATHS) > 1
 _SERIAL_ONLY = "one usable core, so scan_paths stays serial and there is no fan-out to compare"
 

tests/test_litellm/test_test_quality_gate.py

@@ -142,7 +142,14 @@ def test_introduced_keeps_only_violations_on_changed_lines():
 
 def test_the_shipped_budget_covers_every_rule_the_checker_can_emit():
     import json
+    import re
 
+    emitted = set(
+        re.findall(r'"(TQ\d+)"', (_REPO_ROOT / "scripts" / "check_test_quality.py").read_text())
+    )
+    # TQ000 is the unreadable-file/syntax-error code, a hard failure rather than a
+    # countable violation, so it is the one code the budget deliberately omits.
     budget = json.loads((_REPO_ROOT / "test-quality-budget.json").read_text())
-    assert set(budget) == {"TQ001", "TQ002", "TQ003", "TQ004", "TQ005", "TQ006", "TQ007", "TQ008"}
+    assert "TQ000" in emitted
+    assert set(budget) == emitted - {"TQ000"}
     assert all(spec["limit"] >= 0 for spec in budget.values())

tests/test_litellm/test_utils.py

@@ -1922,14 +1922,15 @@ class TestProxyFunctionCalling:
         for model_name, expected_result in test_cases:
             try:
                 result = supports_function_calling(model=model_name)
-                # For malformed models, we expect False or the function to handle gracefully
-                assert (
-                    result == expected_result
-                ), f"Edge case {model_name} returned {result}, expected {expected_result}"
             except Exception:
                 # It's acceptable for malformed model names to raise exceptions
                 # rather than returning False, as long as they're handled gracefully
-                pass
+                continue
+
+            # For malformed models, we expect False or the function to handle gracefully
+            assert (
+                result == expected_result
+            ), f"Edge case {model_name} returned {result}, expected {expected_result}"
 
     def test_proxy_model_resolution_demonstration(self):
         """
@@ -2131,6 +2132,11 @@ class TestProxyFunctionCalling:
         # Test the underlying model directly to verify it supports function calling
         try:
             underlying_result = supports_function_calling(underlying_bedrock_model)
+        except Exception as e:
+            print(
+                f"  Warning: Could not test underlying model {underlying_bedrock_model}: {e}"
+            )
+        else:
             print(f"  Underlying model function calling support: {underlying_result}")
 
             # Most Bedrock Converse API models with Anthropic Claude should support function calling
@@ -2138,10 +2144,6 @@ class TestProxyFunctionCalling:
                 assert (
                     underlying_result is True
                 ), f"Claude 3 models should support function calling: {underlying_bedrock_model}"
-        except Exception as e:
-            print(
-                f"  Warning: Could not test underlying model {underlying_bedrock_model}: {e}"
-            )
 
         # Test the proxy model - should return False due to lack of configuration context
         proxy_result = supports_function_calling(proxy_model_name)
@@ -2224,13 +2226,15 @@ class TestProxyFunctionCalling:
         for model in bedrock_models:
             try:
                 result = supports_function_calling(model)
-                print(f"Direct test - {model}: {result}")
-                # Claude 3 models should support function calling
-                assert (
-                    result is True
-                ), f"Claude 3 model should support function calling: {model}"
             except Exception as e:
                 print(f"Could not test {model}: {e}")
+                continue
+
+            print(f"Direct test - {model}: {result}")
+            # Claude 3 models should support function calling
+            assert (
+                result is True
+            ), f"Claude 3 model should support function calling: {model}"
 
 
 def test_register_model_with_scientific_notation():