test(integration): repair stale contract fixtures (#49700)

tests/integration/caching/test_response_cache_empty_output.py

@@ -23,7 +23,7 @@ from typing import Final, TypeAlias, TypeVar
 import httpx
 import pytest
 from anthropic import Anthropic, AsyncAnthropic
-from openai import AsyncOpenAI, OpenAI
+from openai import AsyncOpenAI, InternalServerError, OpenAI
 from pydantic import JsonValue
 
 from tests.integration._support.client import JSON_OBJECT, Gateway, eventually, object_value, string_value
@@ -420,14 +420,25 @@ def _streamed_chat(gateway: Gateway, model: str, text: str) -> _Answer:
     return ids, streamed, tuple(chunk.object for chunk in chunks)
 
 
-def test_streamed_chat_without_choices_is_never_assembled_or_cached(gateway: Gateway) -> None:
+def test_streamed_chat_without_choices_fails_and_is_never_cached(gateway: Gateway) -> None:
     upstream: Final = Upstream(gateway.upstream_url)
+    store: Final = redis_store()
     text: Final = prompt()
     with gateway.scenario() as scenario:
         handle: Final = scripted(scenario, CHAT_STREAM_WITHOUT_CHOICES)
         model: Final = scenario.model(api_base=handle.api_base())
-        empty: Final = _never_replayed(partial(_streamed_chat, gateway, model, text), upstream, handle.scenario_id)
-        assert all(answer.text == "" for answer in empty), empty
+        stream: Final = partial(_streamed_chat, gateway, model, text)
+        for attempt in range(1, EMPTY_STREAMS + 1):
+            with pytest.raises(InternalServerError, match="Stream ended before a finish_reason was received"):
+                stream()
+            _assert_prompt_reached_upstream(upstream, handle.scenario_id, text, attempt)
+        rescript(handle, CHAT_STREAM)
+        real_ids, real_text, _ = stream()
+        assert real_text == "streamed response" and len(real_ids) == 1, (real_ids, real_text)
+        _assert_prompt_reached_upstream(upstream, handle.scenario_id, text, EMPTY_STREAMS + 1)
+        await_entry(store, next(iter(real_ids)))
+        replay: Final = _replayed(stream, upstream, handle.scenario_id, real_ids)
+        assert replay.text == real_text, replay
 
 
 def _streamed_text(gateway: Gateway, model: str, text: str) -> _Answer:

tests/integration/compatibility/test_a2a_wire_versions.py

@@ -1023,9 +1023,18 @@ def test_a2a_task_methods_pass_through_verbatim_and_push_callbacks_are_validated
                 {"taskId": "t1", "url": callback},
             ):
                 rejected = call("tasks/pushNotificationConfig/set", params)
-                assert (rejected.status_code, rejected.json()) == (400, {"detail": detail}), (
-                    f"{params}: {rejected.text}"
-                )
+                assert (rejected.status_code, rejected.json()) == (
+                    400,
+                    {
+                        "jsonrpc": "2.0",
+                        "id": marker + "tasks/pushNotificationConfig/set",
+                        "error": {
+                            "code": -32000,
+                            "message": detail,
+                            "data": {"type": "invalid_request_error", "param": None, "code": "400"},
+                        },
+                    },
+                ), f"{params}: {rejected.text}"
         assert tuple(item for item in wire.drain() if item.method == "POST") == ()
 
         public: Final = {"taskId": "t1", "pushNotificationConfig": {"url": "https://1.1.1.1/callback"}}
@@ -1118,7 +1127,15 @@ def test_a2a_v1_task_methods_and_extended_card_lower_to_0_3_upstream_and_answer_
             rejected = call("CreateTaskPushNotificationConfig", params)
             assert (rejected.status_code, rejected.json()) == (
                 400,
-                {"detail": "Push notification URL must use HTTPS"},
+                {
+                    "jsonrpc": "2.0",
+                    "id": marker + "CreateTaskPushNotificationConfig",
+                    "error": {
+                        "code": -32000,
+                        "message": "Push notification URL must use HTTPS",
+                        "data": {"type": "invalid_request_error", "param": None, "code": "400"},
+                    },
+                },
             ), f"{params}: {rejected.text}"
         assert forwarded() == ()
 

tests/integration/cost_calculation/cases/azure.py

@@ -1588,6 +1588,7 @@ AZURE_PINNED_GPT_5_4_MINI_STREAM: Final = CostTrackingTestCase(
         content_type="text/event-stream",
         frames=(
             'data: {"id":"chatcmpl-$REQUEST_ID","object":"chat.completion.chunk","created":1,"model":"$MODEL","choices":[{"index":0,"delta":{"role":"assistant","content":"ok"},"finish_reason":null}]}\n\n',
+            'data: {"id":"chatcmpl-$REQUEST_ID","object":"chat.completion.chunk","created":1,"model":"$MODEL","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\n',
             'data: {"id":"chatcmpl-$REQUEST_ID","object":"chat.completion.chunk","created":1,"model":"$MODEL","choices":[],"usage":{"prompt_tokens":1840,"completion_tokens":412,"total_tokens":2252}}\n\n',
             "data: [DONE]\n\n",
         ),

tests/integration/cost_calculation/cases/openai.py

@@ -7952,6 +7952,7 @@ GPT_5_6_STREAM_USAGE_IN_EMPTY_CHOICES_CHUNK: Final = CostTrackingTestCase(
         frames=(
             'data: {"id":"chatcmpl-$REQUEST_ID","object":"chat.completion.chunk","created":1789788263,"model":"gpt-5.6","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}],"usage":null}',
             'data: {"id":"chatcmpl-$REQUEST_ID","object":"chat.completion.chunk","created":1789788263,"model":"gpt-5.6","choices":[{"index":0,"delta":{"content":"scripted answer"},"finish_reason":null}],"usage":null}',
+            'data: {"id":"chatcmpl-$REQUEST_ID","object":"chat.completion.chunk","created":1789788263,"model":"gpt-5.6","choices":[{"index":0,"delta":{},"finish_reason":"stop"}],"usage":null}',
             'data: {"id":"chatcmpl-$REQUEST_ID","object":"chat.completion.chunk","created":1789788263,"model":"gpt-5.6","choices":[],"usage":{"prompt_tokens":1840,"completion_tokens":412,"total_tokens":2252}}',
             "data: [DONE]",
         ),

tests/integration/messages_endpoint/providers/anthropic/test_pdf_document_count_tokens_wire.py

@@ -199,14 +199,14 @@ def test_utils_token_counter_adds_title_and_context_once_per_document(
 
 
 @pytest.mark.parametrize("label", list(_UNREADABLE))
-def test_utils_token_counter_prices_unreadable_pdf_bytes_like_one_image(
+def test_utils_token_counter_uses_the_same_bounded_fallback_for_unreadable_pdf_bytes(
     gateway: Gateway, provider: SharedProvider, label: str
 ) -> None:
     data: Final = _UNREADABLE[label]
     as_pdf: Final = _local(gateway, provider, [document(base64_source(data))])
-    as_png: Final = _local(gateway, provider, [document(base64_source(data, media_type="image/png"))])
+    fallback: Final = _local(gateway, provider, [document(base64_source(encoded(b"unreadable-document")))])
     without: Final = _local(gateway, provider, [])
-    assert as_pdf == as_png, (as_pdf, as_png)
+    assert as_pdf == fallback, (as_pdf, fallback)
     assert 0 <= as_pdf - without < rendered_tokens((LETTER,)), (as_pdf, without)
 
 

tests/integration/observability/test_prometheus_provider_auth_failure.py

@@ -995,7 +995,7 @@ def _gated(gateway: Gateway, cell: str, key: str | None) -> httpx.Response:
 
 
 def recorded_before(cell: str) -> list[tuple[str, str, str, float]]:
-    if cell in ("invalid-key", "expired-key"):
+    if cell in ("invalid-key", "missing-key", "expired-key", "blocked-key"):
         return []
     return sorted([(FAILURES, "", "", 1.0), (TOTALS, "", "", 1.0), (LLM_FAILED, identity(cell), "", 1.0)])
 
@@ -1020,7 +1020,9 @@ def test_a_request_the_proxy_rejects_itself_never_marks_the_deployment_down_and_
         assert response.status_code == (403 if cell == "no-model-access" else 401), response.text
         assert rig.upstream.drain() == ()
         samples: Final = until(rig.gateway, PROXY_FAILED, requested_model=name(cell))
+        assert values(samples, PROXY_FAILED, requested_model=name(cell)) == [1.0], recorded(samples)
         assert deployment_rows(samples, cell) == recorded_before(cell), recorded(samples, requested_model=name(cell))
+        assert values(samples, STATE, model_id=identity(cell)) == [], recorded(samples)
 
 
 def test_a_key_rate_limit_rejected_at_the_proxy_is_still_counted_as_before(rig: Rig) -> None:

tests/integration/providers/test_device_code_login_guard_boot.py

@@ -28,7 +28,6 @@ from pydantic import JsonValue, TypeAdapter
 pytestmark: Final = [pytest.mark.timeout(900), pytest.mark.shares_worker_state]
 
 _STARTED_WORKER: Final = re.compile(r"Started server process \[(\d+)\]")
-_DROPPED: Final = re.compile(r"original model: (\S+), ignoring and continuing")
 _FIXED: Final[Mapping[str, str]] = MappingProxyType(
     {"chatgpt-fixed": "chatgpt/gpt-5.6-terra", "copilot-fixed": "github_copilot/gpt-5.4"}
 )
@@ -41,8 +40,6 @@ _CONTROL_CALLS_BEFORE_THE_RELOAD: Final = 1
 _SHAPES: Final[Mapping[str, bool]] = MappingProxyType(
     {"docs-shapes": False, "docs-shapes-beside-a-universal-wildcard": True}
 )
-_NOT_FOUND: Final = "Invalid model name passed in model="
-_NO_HEALTHY: Final = "There are no healthy deployments for this model"
 _CONTROL: Final = "device-login-boot-control"
 _EXTRA: Final[Mapping[str, JsonValue]] = MappingProxyType({"num_retries": 0, "cache": {"no-cache": True}})
 _CALL_ID: Final = "x-litellm-call-id"
@@ -174,15 +171,11 @@ def _lists_login_models(listed: tuple[str, ...]) -> bool:
     )
 
 
-def _every_worker_lists(
-    owned: OwnedProxy, workers: tuple[int, ...], *, login_models: bool
-) -> Mapping[int, tuple[str, ...]]:
+def _every_worker_lists(owned: OwnedProxy, workers: tuple[int, ...]) -> Mapping[int, tuple[str, ...]]:
     def settled(by_worker: Mapping[int, tuple[str, ...]]) -> bool:
         if set(by_worker) != set(workers):
             return False
-        if login_models:
-            return all(_lists_login_models(listed) for listed in by_worker.values())
-        return all(_login_names_listed(listed) == frozenset() for listed in by_worker.values())
+        return all(_lists_login_models(listed) for listed in by_worker.values())
 
     return eventually(lambda: _listing_by_worker(owned, workers), settled, seconds=90)
 
@@ -212,16 +205,8 @@ def _reload(owned: OwnedProxy) -> None:
     assert response.status_code == 200, response.text
 
 
-def _refusal_before_the_reload(name: str, universal: bool) -> str:
-    if name in _FIXED:
-        return _NO_HEALTHY
-    if not universal:
-        return _NOT_FOUND
-    return dl.CHATGPT_REFUSAL if name.startswith("chatgpt/") else dl.COPILOT_REFUSAL
-
-
 @pytest.mark.parametrize("universal", _SHAPES.values(), ids=_SHAPES.keys())
-def test_booting_without_tokens_drops_the_login_deployments_and_a_reload_after_mounting_brings_them_back(
+def test_booting_without_tokens_lists_login_deployments_and_serves_only_after_tokens_are_mounted(
     gateway: Gateway, tmp_path: Path, universal: bool
 ) -> None:
     dirs: Final = _token_dirs(tmp_path)
@@ -240,20 +225,18 @@ def test_booting_without_tokens_drops_the_login_deployments_and_a_reload_after_m
             workers=2,
         ) as owned:
             workers: Final = _workers(owned)
-            _every_worker_lists(owned, workers, login_models=False)
+            _every_worker_lists(owned, workers)
             boot_log: Final = owned.log.read_text()
             assert peers.auth_connections() == ()
             assert dl.CHATGPT_USER_CODE not in boot_log and dl.COPILOT_USER_CODE not in boot_log, boot_log[-3000:]
-            dropped_models: Final = _DROPPED.findall(boot_log)
-            expected_drops: Final = frozenset((*_FIXED.values(), *_WILDCARDS))
-            assert frozenset(dropped_models) == expected_drops, boot_log[-3000:]
-            assert all(dropped_models.count(model) >= len(workers) for model in expected_drops), boot_log[-3000:]
             with owned.gateway.scenario() as scenario:
                 key: Final = scenario.key()
                 for name in _LOGIN_NAMES:
                     refused: Final = _chat(_base_url(owned), key, name, uuid.uuid4().hex)
                     assert refused.status == 400, refused.text
-                    assert _refusal_before_the_reload(name, universal) in refused.message(), refused.text
+                    refusal: Final = dl.CHATGPT_REFUSAL if name.startswith("chatgpt") else dl.COPILOT_REFUSAL
+                    assert refusal in refused.message(), refused.text
+                assert peers.api.drain() == ()
                 control: Final = _chat(_base_url(owned), key, _CONTROL, uuid.uuid4().hex)
                 assert control.status == 200, control.text
                 assert peers.auth_connections() == ()
@@ -261,7 +244,7 @@ def test_booting_without_tokens_drops_the_login_deployments_and_a_reload_after_m
 
                 dirs.store_valid_tokens(peers.api.url)
                 _reload(owned)
-                _every_worker_lists(owned, workers, login_models=True)
+                _every_worker_lists(owned, workers)
                 for served_logins, name in enumerate(_LOGIN_NAMES, start=1):
                     served: Final = _chat(_base_url(owned), key, name, uuid.uuid4().hex)
                     assert served.status == 200, served.text

tests/integration/providers/test_device_code_login_guard_wire.py

@@ -746,10 +746,7 @@ def test_a_recent_device_code_request_does_not_hold_the_worker_for_the_rest_of_t
     assert scene.rig.peers.auth_connections() == ()
 
 
-_NOT_LIVE: Final = re.compile(r"saved to the database, but the model id\(s\) \['([0-9a-f-]+)'\] are not live")
-
-
-def test_model_created_without_a_token_is_saved_not_live_and_serves_once_the_token_is_mounted(scene: _Scene) -> None:
+def test_model_created_without_a_token_refuses_requests_and_serves_once_the_token_is_mounted(scene: _Scene) -> None:
     gateway: Final = scene.rig.owned.gateway
     name: Final = f"device-login-db-{uuid.uuid4().hex}"
     identity: Final = str(uuid.uuid4())
@@ -763,14 +760,18 @@ def test_model_created_without_a_token_is_saved_not_live_and_serves_once_the_tok
         },
     )
     try:
-        assert created.status_code == 500, created.text
-        not_live: Final = _NOT_LIVE.search(created.text)
-        assert not_live is not None and not_live.group(1) == identity, created.text
+        assert created.status_code == 200, created.text
+        assert created.json()["model_info"]["id"] == identity, created.text
         assert scene.rig.peers.auth_connections() == ()
-        scene.store_valid_token("chatgpt")
         call: Final = _Call(
             scene.rig.base_url(), scene.key, name, "responses", False, "httpx", uuid.uuid4().hex, SimpleQueue()
         )
+        refused: Final = _serve(call)
+        assert refused.status == 400, refused.text
+        assert dl.CHATGPT_REFUSAL in _error_message(refused.text), refused.text
+        assert scene.rig.peers.auth_connections() == ()
+        assert scene.rig.peers.api.drain() == ()
+        scene.store_valid_token("chatgpt")
         served: Final = eventually(lambda: _serve(call), lambda answer: answer.status == 200, seconds=90)
         assert set(rv.MARKER.findall(served.text)) == {call.marker}, served.text
         assert scene.rig.peers.auth_connections() == ()

tests/integration/sdk/test_anthropic_json_tool_cut_stream_sync_sdk.py

@@ -10,6 +10,7 @@ from integration._support.wire import Reply, Request, wire_server
 from pydantic import JsonValue
 
 import litellm
+from litellm.exceptions import MidStreamFallbackError
 from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
 
 _JSON_TOOL: Final = "json_tool_call"
@@ -122,12 +123,10 @@ def _credentials(model: str) -> dict[str, str]:
 @pytest.mark.parametrize("model", [_ANTHROPIC_MODEL, _BEDROCK_INVOKE_MODEL], ids=["anthropic", "bedrock_invoke"])
 @pytest.mark.parametrize(
     ("shape", "cut"),
-    [("plain", False), ("plain", True), ("wrapped", True)],
-    ids=["plain_complete", "plain_cut", "wrapped_cut"],
+    [("plain", False), ("wrapped", False), ("plain", True), ("wrapped", True)],
+    ids=["plain_complete", "wrapped_complete", "plain_cut", "wrapped_cut"],
 )
-def test_sync_stream_cut_before_the_json_tool_block_closes_still_yields_the_whole_answer(
-    model: str, shape: str, cut: bool
-) -> None:
+def test_sync_json_tool_stream_requires_a_provider_finish_reason(model: str, shape: str, cut: bool) -> None:
     marker_hex: Final = uuid.uuid4().hex
     with wire_server(_peer(marker_hex, shape, cut)) as wire:
         response: Final = litellm.completion(
@@ -141,6 +140,11 @@ def test_sync_stream_cut_before_the_json_tool_block_closes_still_yields_the_whol
             **_credentials(model),
         )
         assert isinstance(response, CustomStreamWrapper), type(response)
+        if cut:
+            with pytest.raises(MidStreamFallbackError, match="Stream ended before a finish_reason was received"):
+                tuple(response)
+            assert len(wire.drain()) == 1
+            return
         text: Final = "".join(str(chunk.choices[0].delta.content or "") for chunk in tuple(response))
         assert len(wire.drain()) == 1
     assert json.loads(text) == _payload(marker_hex), text