chore(proxy): extract _is_json_content_type symmetric helper

Mirror ``_is_form_content_type`` for the JSON branch of
``get_request_body`` so both classifications share the same media-type
normalisation (strip params, trim, lowercase) and any future change
to the parsing rules has one place to update.

Adds tests for ``_is_json_content_type`` and for ``get_request_body``
covering the canonical JSON / form / unsupported / non-POST paths.
This commit is contained in:
user 2026-05-14 14:52:04 +00:00
parent 2ec54106e8
commit 56eb5ef550
No known key found for this signature in database
2 changed files with 85 additions and 5 deletions

View file

@ -18,6 +18,13 @@ _FORM_CONTENT_TYPES: frozenset[str] = frozenset(
)
def _normalize_media_type(content_type: str) -> str:
"""Return the bare media type per RFC 7231: strip params, trim, lowercase."""
if not content_type:
return ""
return content_type.split(";", 1)[0].strip().lower()
def _is_form_content_type(content_type: str) -> bool:
"""
True iff Starlette's ``request.form()`` will actually parse this body.
@ -26,10 +33,12 @@ def _is_form_content_type(content_type: str) -> bool:
``FormData`` for non-canonical types without consuming the body, leaving
the auth-time pre-read and the handler's read seeing different payloads.
"""
if not content_type:
return False
media_type = content_type.split(";", 1)[0].strip().lower()
return media_type in _FORM_CONTENT_TYPES
return _normalize_media_type(content_type) in _FORM_CONTENT_TYPES
def _is_json_content_type(content_type: str) -> bool:
"""True iff the body should be parsed as JSON."""
return _normalize_media_type(content_type) == "application/json"
async def _read_request_body(request: Optional[Request]) -> Dict:
@ -326,7 +335,7 @@ async def get_request_body(request: Request) -> Dict[str, Any]:
"""
if request.method == "POST":
content_type = request.headers.get("content-type", "")
if content_type.split(";", 1)[0].strip().lower() == "application/json":
if _is_json_content_type(content_type):
return await _read_request_body(request)
elif _is_form_content_type(content_type):
return await get_form_data(request)

View file

@ -17,6 +17,7 @@ import litellm
from litellm.proxy._types import ProxyException
from litellm.proxy.common_utils.http_parsing_utils import (
_is_form_content_type,
_is_json_content_type,
_read_request_body,
_safe_get_request_headers,
_safe_get_request_parsed_body,
@ -927,3 +928,73 @@ class TestReadRequestBodyNonCanonicalContentType:
result = await _read_request_body(mock_request)
assert result == {"k": "v"}
mock_request.form.assert_awaited_once()
class TestIsJsonContentType:
@pytest.mark.parametrize(
"content_type",
[
"application/json",
"application/json; charset=utf-8",
"Application/JSON",
" application/json ",
],
)
def test_json_types_match(self, content_type):
assert _is_json_content_type(content_type) is True
@pytest.mark.parametrize(
"content_type",
[
"",
"application/x-www-form-urlencoded",
"multipart/form-data",
"application/jsonl",
"application/jsonschema",
"text/json",
"application/json-patch+json",
],
)
def test_non_json_types_rejected(self, content_type):
assert _is_json_content_type(content_type) is False
class TestGetRequestBody:
@pytest.mark.asyncio
async def test_json_with_charset_param_parses_as_json(self):
payload = {"k": "v"}
mock_request = MagicMock()
mock_request.method = "POST"
mock_request.body = AsyncMock(return_value=orjson.dumps(payload))
mock_request.headers = {"content-type": "application/json; charset=utf-8"}
mock_request.scope = {}
result = await get_request_body(mock_request)
assert result == payload
@pytest.mark.asyncio
async def test_form_post_routes_to_form_data(self):
mock_request = MagicMock()
mock_request.method = "POST"
mock_request.headers = {"content-type": "multipart/form-data; boundary=x"}
mock_request.form = AsyncMock(return_value={"k": "v"})
mock_request.scope = {}
result = await get_request_body(mock_request)
assert result == {"k": "v"}
@pytest.mark.asyncio
async def test_substring_match_no_longer_accepted(self):
mock_request = MagicMock()
mock_request.method = "POST"
mock_request.headers = {"content-type": "application/form-json"}
mock_request.scope = {}
with pytest.raises(ValueError, match="Unsupported content type"):
await get_request_body(mock_request)
@pytest.mark.asyncio
async def test_non_post_returns_empty(self):
mock_request = MagicMock()
mock_request.method = "GET"
assert await get_request_body(mock_request) == {}