scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-03 02:22:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 80

test(s3_v2): pin async 5xx retry through the production AsyncHTTPHandler (#43080)

Browse source 
* test(s3_v2): pin async 5xx retry through the production AsyncHTTPHandler

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* Update tests/unit/integrations/test_s3_v2.py

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

---------

Co-authored-by: shivam <shivam@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Mrinal Chanshetty <mrinal@berri.ai>
Co-authored-by: yucheng-berri <yucheng@berri.ai>
Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-09-30 14:30:27 -07:00 • committed by GitHub
parent e61733b170
commit cbe69723b1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 30 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
tests/unit/integrations/test_s3_v2.py
View file

@ -672,6 +672,36 @@ async def test_async_upload_exhausts_403_retries_through_production_http_handler
assert "Error uploading to s3" in caplog.text
@pytest.mark.asyncio
@pytest.mark.parametrize("transient_status", [500, 503])
async def test_async_upload_recovers_from_transient_5xx_through_production_http_handler(
transient_status: int, rotating_profile: str, caplog: pytest.LogCaptureFixture
):
"""
AsyncHTTPHandler.put raises MaskedHTTPStatusError on 5xx instead of returning the response, so a retry
loop that only inspects returned status codes never runs (#42868).
"""
test_element = s3BatchLoggingElement(
s3_object_key=f"2025-09-14/test-{transient_status}.json",
payload={"test": str(transient_status)},
s3_object_download_filename=f"test-{transient_status}.json",
)
async with _s3_logger_on_production_handler(rotating_profile, [transient_status, 200]) as (
logger,
requests,
mock_sleep,
):
uploaded = await logger.async_upload_data_to_s3(test_element)
assert uploaded is True
assert len(requests) == 2
assert all(request.method == "PUT" for request in requests)
assert requests[0].url == requests[1].url
assert requests[0].content == requests[1].content
mock_sleep.assert_awaited_once_with(1)
assert "Error uploading to s3" not in caplog.text
@pytest.mark.asyncio
async def test_async_upload_is_single_attempted_on_404_through_production_http_handler(rotating_profile: str, caplog):
test_element = s3BatchLoggingElement(