From 4cc9f561bc2306fa98518c5083e19090ab968d47 Mon Sep 17 00:00:00 2001 From: yucheng-berri Date: Thu, 8 Oct 2026 16:31:54 -0700 Subject: [PATCH] test(batches): pin the all-or-nothing retry and warm the burst model first An unreadable error file emits no lines and the next retrieve emits all of them. The burst cell drives one speech model after a warm-up request so a cold worker cannot fail it, and waits for the batched key spend. --- .../spend/test_batch_parent_id_body_field.py | 19 ++++++----- .../batches/test_batch_line_item_logging.py | 34 +++++++++++++++++++ 2 files changed, 44 insertions(+), 9 deletions(-) diff --git a/tests/integration/spend/test_batch_parent_id_body_field.py b/tests/integration/spend/test_batch_parent_id_body_field.py index c5502e4957d..d99d68d033f 100644 --- a/tests/integration/spend/test_batch_parent_id_body_field.py +++ b/tests/integration/spend/test_batch_parent_id_body_field.py @@ -237,17 +237,18 @@ def test_any_shape_of_batch_parent_id_on_a_speech_request_is_still_tracked( def test_a_concurrent_burst_mixing_forged_and_plain_requests_bills_every_request(gateway: Gateway) -> None: with gateway.scenario() as scenario: - models: Final = tuple( - _scripted_model(gateway, scenario, "openai/omni-moderation-latest", _moderation_reply) for _ in range(20) - ) - key: Final = scenario.key(models=list(models)) + model: Final = _scripted_model(gateway, scenario, "openai/tts-1", _speech_reply) + key: Final = scenario.key(models=[model]) + warm_up: Final = _speech(gateway, model, key, {}) + assert warm_up.status_code == 200, warm_up.text + eventually(lambda: _spend_rows(key), lambda values: len(values) == 1, seconds=70) extras: Final = tuple({"batch_parent_id": FORGED_PARENT} if index % 2 else {} for index in range(20)) with ThreadPoolExecutor(max_workers=10) as pool: - responses: Final = tuple( - pool.map(lambda pair: _moderation(gateway, pair[0], key, pair[1]), zip(models, extras)) - ) + responses: Final = tuple(pool.map(lambda extra: _speech(gateway, model, key, extra), extras)) assert [response.status_code for response in responses] == [200] * 20, [r.text for r in responses] - rows: Final = eventually(lambda: _spend_rows(key), lambda values: len(values) == 20, seconds=90) - assert {row["call_type"] for row in rows} == {"amoderation"}, rows + rows: Final = eventually(lambda: _spend_rows(key), lambda values: len(values) == 21, seconds=90) + assert {row["call_type"] for row in rows} == {"aspeech"}, rows + expected_spend: Final = 21 * float(str(rows[0]["spend"])) + eventually(lambda: _key_spend(key), lambda value: value == pytest.approx(expected_spend), seconds=90) diff --git a/tests/unit/batches/test_batch_line_item_logging.py b/tests/unit/batches/test_batch_line_item_logging.py index d383546b7bf..f014c6e6fde 100644 --- a/tests/unit/batches/test_batch_line_item_logging.py +++ b/tests/unit/batches/test_batch_line_item_logging.py @@ -1241,6 +1241,40 @@ async def test_supplied_result_files_skip_the_output_and_error_fetch(recorder): assert file_mock.await_args.kwargs["file_id"] == "input-file-1" +def _error_file_unreadable(file_id: str, **_kwargs): + if file_id == "error-file-1": + raise ValueError("error file gone") + return SimpleNamespace(content=_FILE_BYTES[file_id]) + + +@pytest.mark.asyncio +async def test_an_unreadable_error_file_defers_every_line_to_the_next_retrieve(recorder): + claim_cache: Final = DualCache() + batch: Final = _batch() + parent: Final = _parent_logging() + emitted: list[int] = [] # mutable-ok: collects one count per retrieve + for side_effect in (_error_file_unreadable, _file_content): + with patch( + "litellm.files.main.afile_content", AsyncMock(side_effect=side_effect) + ): # test-quality-ok: afile_content is the provider boundary; no injection seam for managed file fetch + emitted.append( + await log_batch_line_items( + batch=batch, + custom_llm_provider="openai", + parent=parent, + model_name="gpt-4o", + litellm_params=None, + model_info=None, + result_files=BatchResultFiles(output=OUTPUT_JSONL, error=None), + claim_cache=claim_cache, + ) + ) + + assert emitted == [0, 2] + assert len(recorder.success_events) == 1 + assert len(recorder.failure_events) == 1 + + @pytest.mark.asyncio async def test_empty_result_files_fall_back_to_fetching_everything(recorder): file_mock: Final = AsyncMock(side_effect=_file_content)