scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-05 08:07:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 137

feat(ci): fail the build on a test that catches its own AssertionError

Browse source 
A failing `assert` raises AssertionError, which is an Exception, so an
`except Exception` in the same test catches it, the handler returns, and
pytest records a pass on exactly the regression the assertion guards. TQ009
finds the 29 tests in the suite shaped that way and seeds the ceiling at the
23 left after this branch.

Six of them were in tests/test_litellm/. Narrowing the try to the call and
moving the assertions below it turned two of them red:

- test_sentry_sample_rate asserted SENTRY_API_SAMPLE_RATE was written back
  to os.environ, which set_callbacks never does. It now reads the rate off
  the initialised client, which is where the value actually lands.
- test_convert_tool_response_with_url_image downloaded from a host that no
  longer resolves, and the product swallows a fetch failure, so the skip the
  author wrote could never fire and the assertion failure it caught instead
  read as a network skip. The download is now served from a stubbed
  transport, so the media path runs offline and deterministically.

Also stops two sentry tests leaking a MagicMock into sys.modules, and derives
the budget-coverage test's rule set from the checker instead of a literal
list that goes stale on every new rule.
This commit is contained in:
ryan-crabbe-berri 2026-08-26 18:50:43 -07:00
parent 77765fd302
commit ee4547d41f
8 changed files with 358 additions and 79 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

74
scripts/check_test_quality.py
View file

@ -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
@ -134,6 +145,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 +419,66 @@ 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 _reports_the_failure(handler: ast.ExceptHandler) -> bool:
"""Re-raising, or failing the test, passes the failure on rather than eating it."""
return any(
isinstance(node, ast.Raise)
or (isinstance(node, ast.Call) and _is_pytest_assertion_call(node))
for parent in handler.body
for node in ast.walk(parent)
) or _raises_assertion_error(handler.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)
),
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 +813,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),

3
test-quality-budget.json
View file

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

38
tests/test_litellm/litellm_core_utils/test_litellm_logging.py
View file

@ -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"])

63
tests/test_litellm/llms/vertex_ai/gemini/test_vertex_ai_gemini_transformation.py
View file

@ -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():

20
tests/test_litellm/proxy/pass_through_endpoints/test_pass_through_endpoints.py
View file

@ -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():

198
tests/test_litellm/test_check_test_quality.py
View file

@ -681,6 +681,204 @@ 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_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"

9
tests/test_litellm/test_test_quality_gate.py
View file

@ -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())

32
tests/test_litellm/test_utils.py
View file

@ -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():