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)