forked from phoenix/litellm-mirror
21156ff5d0
* fix(http_parsing_utils.py): remove `ast.literal_eval()` from http utils Security fix - https://huntr.com/bounties/96a32812-213c-4819-ba4e-36143d35e95b?token=bf414bbd77f8b346556e 64ab2dd9301ea44339910877ea50401c76f977e36cdd78272f5fb4ca852a88a7e832828aae1192df98680544ee24aa98f3cf6980d8 bab641a66b7ccbc02c0e7d4ddba2db4dbe7318889dc0098d8db2d639f345f574159814627bb084563bad472e2f990f825bff0878a9 e281e72c88b4bc5884d637d186c0d67c9987c57c3f0caf395aff07b89ad2b7220d1dd7d1b427fd2260b5f01090efce5250f8b56ea2 c0ec19916c24b23825d85ce119911275944c840a1340d69e23ca6a462da610 * fix(converse/transformation.py): support bedrock apac cross region inference Fixes https://github.com/BerriAI/litellm/issues/6905 * fix(user_api_key_auth.py): add auth check for websocket endpoint Fixes https://github.com/BerriAI/litellm/issues/6926 * fix(user_api_key_auth.py): use `model` from query param * fix: fix linting error * test: run flaky tests first
79 lines
2.2 KiB
Python
79 lines
2.2 KiB
Python
import pytest
|
|
from fastapi import Request
|
|
from fastapi.testclient import TestClient
|
|
from starlette.datastructures import Headers
|
|
from starlette.requests import HTTPConnection
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(
|
|
0, os.path.abspath("../..")
|
|
) # Adds the parent directory to the system path
|
|
|
|
from litellm.proxy.common_utils.http_parsing_utils import _read_request_body
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_request_body_valid_json():
|
|
"""Test the function with a valid JSON payload."""
|
|
|
|
class MockRequest:
|
|
async def body(self):
|
|
return b'{"key": "value"}'
|
|
|
|
request = MockRequest()
|
|
result = await _read_request_body(request)
|
|
assert result == {"key": "value"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_request_body_empty_body():
|
|
"""Test the function with an empty body."""
|
|
|
|
class MockRequest:
|
|
async def body(self):
|
|
return b""
|
|
|
|
request = MockRequest()
|
|
result = await _read_request_body(request)
|
|
assert result == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_request_body_invalid_json():
|
|
"""Test the function with an invalid JSON payload."""
|
|
|
|
class MockRequest:
|
|
async def body(self):
|
|
return b'{"key": value}' # Missing quotes around `value`
|
|
|
|
request = MockRequest()
|
|
result = await _read_request_body(request)
|
|
assert result == {} # Should return an empty dict on failure
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_request_body_large_payload():
|
|
"""Test the function with a very large payload."""
|
|
large_payload = '{"key":' + '"a"' * 10**6 + "}" # Large payload
|
|
|
|
class MockRequest:
|
|
async def body(self):
|
|
return large_payload.encode()
|
|
|
|
request = MockRequest()
|
|
result = await _read_request_body(request)
|
|
assert result == {} # Large payloads could trigger errors, so validate behavior
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_request_body_unexpected_error():
|
|
"""Test the function when an unexpected error occurs."""
|
|
|
|
class MockRequest:
|
|
async def body(self):
|
|
raise ValueError("Unexpected error")
|
|
|
|
request = MockRequest()
|
|
result = await _read_request_body(request)
|
|
assert result == {} # Ensure fallback behavior
|