test(integration): repair seven stale integration tests on main (#47600)

The Gemini TTS billing and text completion guardrail tests read spend rows from the owned proxy's own database, which #47213 made private. The cursor wire test derives its effort levels from the gpt-5 cost map row that #43935 now enforces, and the Azure effort check accepts the unquoted level. The pin TTL test asserts the one hour pin #46411 writes for a tool-only mark, the video callback test decodes ids with the product decoder after #44614, and the two Gemini 3.8 flash stream cost cases bill audio at the text rate their rows imply

tests/integration/cost_calculation/cases/gemini.py

@@ -2658,7 +2658,7 @@ GEMINI_GEMINI_3_8_FLASH_STREAM_FULL_USAGE: Final = CostTrackingTestCase(
         ),
     ),
     expected=ExactExpected(
-        spend=0.0076232, input_cost=0.0015572, output_cost=0.006066, prompt_tokens=8314, completion_tokens=1592
+        spend=0.0066182, input_cost=0.0013922, output_cost=0.005226, prompt_tokens=8314, completion_tokens=1592
     ),
 )
 

tests/integration/cost_calculation/cases/vertex_ai.py

@@ -2846,7 +2846,7 @@ GEMINI_3_8_FLASH_STREAM_FULL_USAGE: Final = CostTrackingTestCase(
         ),
     ),
     expected=ExactExpected(
-        spend=0.005775328, input_cost=0.001276288, output_cost=0.00449904, prompt_tokens=8314, completion_tokens=1592
+        spend=0.005946928, input_cost=0.001447888, output_cost=0.00449904, prompt_tokens=8314, completion_tokens=1592
     ),
 )
 

tests/integration/observability/test_callback_delivery.py

@@ -1,4 +1,3 @@
-import base64
 import json
 import uuid
 from collections.abc import Callable, Mapping
@@ -14,6 +13,8 @@ from integration._support.database import read_rows
 from integration._support.process import graceful_stop_seconds, owned_proxy
 from integration._support.wire import Reply, Request, wire_server
 
+from litellm.types.videos.utils import decode_video_id_with_provider
+
 pytestmark: Final = pytest.mark.shares_worker_state
 
 
@@ -431,9 +432,8 @@ def _responses_text(body: dict[str, JsonValue]) -> str:
 
 
 def _video_job(body: dict[str, JsonValue]) -> tuple[str, str]:
-    encoded_id: Final = string_value(body["id"]).removeprefix("video_")
-    decoded: Final = base64.b64decode(encoded_id).decode()
-    return decoded.rsplit("video_id:", 1)[-1], string_value(body["status"])
+    decoded: Final = decode_video_id_with_provider(string_value(body["id"]))
+    return string_value(decoded.get("video_id")), string_value(body["status"])
 
 
 @dataclass(frozen=True, slots=True)

tests/integration/providers/test_azure_reasoning_effort_none_wire.py

@@ -1,3 +1,4 @@
+import re
 from typing import Final
 
 from integration._support.client import Gateway
@@ -30,7 +31,7 @@ def test_reasoning_effort_none_is_rejected_by_litellm_before_reaching_azure(gate
         )
 
         assert response.status_code == 400, response.text
-        assert "reasoning_effort='none'" in response.text, response.text
+        assert re.search(r"reasoning_effort='?none\b", response.text), response.text
         assert wire.drain() == ()
 
 

tests/integration/providers/test_cursor_responses_wire.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import json
 import os
 import uuid
+from importlib.resources import files
 from typing import Final
 
 import pytest
@@ -16,8 +17,16 @@ _MODEL: Final = "gpt-5"
 _API_KEY: Final = "synthetic-cursor-key"
 _JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
 _PROXY_WORKERS: Final = max(1, int(os.environ.get("INTEGRATION_PROXY_WORKERS", "1")))
-_THINKING_LEVELS: Final = ("none", "minimal", "low", "medium", "high", "xhigh", "max")
-_CHAT_THINKING_LEVELS: Final = ("none", "minimal", "low", "medium", "high", "max")
+_COST_MAP: Final = TypeAdapter(dict[str, dict[str, JsonValue]])
+_MODEL_ROW: Final = _COST_MAP.validate_json(
+    (files("litellm") / "model_prices_and_context_window_backup.json").read_bytes()
+)[_MODEL]
+_THINKING_LEVELS: Final = tuple(
+    level
+    for level in ("none", "minimal", "low", "medium", "high", "xhigh", "max")
+    if _MODEL_ROW.get(f"supports_{level}_reasoning_effort") is not False
+)
+_CHAT_THINKING_LEVELS: Final = tuple(level for level in _THINKING_LEVELS if level != "xhigh")
 _RESPONSES_SUFFIXES: Final[tuple[tuple[str, dict[str, JsonValue]], ...]] = tuple(
     (f"-thinking-{level}", {"reasoning": {"effort": level}}) for level in _THINKING_LEVELS
 ) + (

tests/integration/routing/test_prompt_cache_pin_ttl.py

@@ -270,7 +270,7 @@ def test_chat_pin_lifetime_follows_the_longest_breakpoint_ttl_in_the_cacheable_p
     spend_row_lands(response_id)
 
 
-def test_chat_one_hour_mark_on_a_tool_definition_alone_writes_no_pin(rig: _Rig) -> None:
+def test_chat_one_hour_mark_on_a_tool_definition_alone_pins_the_served_deployment_for_an_hour(rig: _Rig) -> None:
     before: Final = _before(rig)
     marker: Final = new_marker()
     tool: Final[dict[str, JsonValue]] = {
@@ -281,8 +281,10 @@ def test_chat_one_hour_mark_on_a_tool_definition_alone_writes_no_pin(rig: _Rig)
     response_id, served = post_chat(
         rig.owned.gateway, chat_body(_CLAUDE, _first_turn(new_marker(), marker, None), tools=[tool])
     )
-    assert served == rig.served_by(marker), served
-    no_pin_written(rig.redis, before, response_id)
+    _, pin = one_new_pin(rig.redis, before)
+    assert served == rig.served_by(marker) == pin.model_id, (served, pin)
+    assert lives_one_hour(pin), pin
+    spend_row_lands(response_id)
 
 
 def _openai_client(rig: _Rig) -> openai.OpenAI:

tests/integration/spend/test_gemini_tts_speech_billing.py

@@ -11,7 +11,6 @@ from typing import Final, Literal
 import pytest
 import yaml
 from integration._support.client import JSON_OBJECT, eventually, gateway_from_environment, object_value
-from integration._support.database import read_rows
 from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
 from integration._support.vertex import service_account_json
 from integration._support.wire import Wire, wire_server
@@ -201,13 +200,15 @@ def _upstream_calls(wire: Wire, text: str) -> int:
     return sum(1 for request in wire.drain() if received_text(request) == text)
 
 
-def _spend_rows(key: str, count: int) -> list[dict[str, JsonValue]]:
-    return eventually(lambda: read_rows(SPEND_ROWS_SQL, (_hashed(key),)), lambda rows: len(rows) == count, seconds=70)
+def _spend_rows(billing: BillingProxy, key: str, count: int) -> list[dict[str, JsonValue]]:
+    return eventually(
+        lambda: billing.proxy.gateway.rows(SPEND_ROWS_SQL, (_hashed(key),)), lambda rows: len(rows) == count, seconds=70
+    )
 
 
-def _key_spend(key: str, expected: float) -> JsonValue:
+def _key_spend(billing: BillingProxy, key: str, expected: float) -> JsonValue:
     rows: Final = eventually(
-        lambda: read_rows(KEY_SPEND_SQL, (_hashed(key),)),
+        lambda: billing.proxy.gateway.rows(KEY_SPEND_SQL, (_hashed(key),)),
         lambda found: len(found) == 1 and found[0]["spend"] == pytest.approx(expected),
         seconds=70,
         return_last_on_timeout=True,
@@ -224,7 +225,7 @@ def _chat_spend(billing: BillingProxy, model: str, key: str) -> float:
     }
     response: Final = billing.proxy.gateway.request("POST", "/v1/chat/completions", body, key=key)
     assert response.status_code == 200, response.text
-    row: Final = _spend_rows(key, 1)[0]
+    row: Final = _spend_rows(billing, key, 1)[0]
     assert (row["prompt_tokens"], row["completion_tokens"]) == (PROMPT_TOKENS, AUDIO_TOKENS), row
     spend: Final = row["spend"]
     assert isinstance(spend, float) and spend > 0, row
@@ -261,11 +262,11 @@ def test_speech_calls_google_once_and_bills_and_logs_its_token_usage_once(
         assert wav_payload(sent.body) == PCM
         assert _upstream_calls(tts_wire, text) == 1
         chat_spend: Final = _chat_spend(uncached, MODELS[provider], scenario.key())
-        row: Final = _spend_rows(speech_key, 1)[0]
+        row: Final = _spend_rows(uncached, speech_key, 1)[0]
         assert row["call_type"] == "aspeech", row
         assert (row["prompt_tokens"], row["completion_tokens"]) == (PROMPT_TOKENS, AUDIO_TOKENS), row
         assert row["spend"] == pytest.approx(chat_spend), row
-        assert _key_spend(speech_key, chat_spend) == pytest.approx(chat_spend)
+        assert _key_spend(uncached, speech_key, chat_spend) == pytest.approx(chat_spend)
         events: Final = _speech_events(uncached, speech_key, 1)
         assert [_summary(event) for event in events] == [("aspeech", BINARY_RESPONSE, PROMPT_TOKENS, AUDIO_TOKENS)]
         assert events[0]["response_cost"] == pytest.approx(chat_spend), events
@@ -286,11 +287,11 @@ def test_cached_speech_repeat_is_a_free_cache_hit_after_a_billed_first_call(
         assert _upstream_calls(tts_wire, text) == 1
         assert repeat.cache_key, "the cached repeat carried no x-litellm-cache-key header"
         chat_spend: Final = _chat_spend(cached, model, scenario.key())
-        billed, hit = _spend_rows(speech_key, 2)
+        billed, hit = _spend_rows(cached, speech_key, 2)
         assert billed["cache_hit"] != "True" and billed["spend"] == pytest.approx(chat_spend), billed
         assert (billed["prompt_tokens"], billed["completion_tokens"]) == (PROMPT_TOKENS, AUDIO_TOKENS), billed
         assert (hit["cache_hit"], hit["spend"]) == ("True", 0.0), hit
-        assert _key_spend(speech_key, chat_spend) == pytest.approx(chat_spend)
+        assert _key_spend(cached, speech_key, chat_spend) == pytest.approx(chat_spend)
         events: Final = _speech_events(cached, speech_key, 2)
         assert len(events) == 2, events
         hits: Final = tuple(event for event in events if event["cache_hit"] is True)
@@ -325,7 +326,7 @@ def test_a_speech_reply_fills_the_deployment_tpm_limit_so_the_next_request_gets_
         first_text: Final = marker()
         first: Final = _speech(uncached, TPM_MODEL, first_text, key)
         assert first.status == 200, first.body[:300]
-        _spend_rows(key, 1)
+        _spend_rows(uncached, key, 1)
         second_text: Final = marker()
         second: Final = _speech(uncached, TPM_MODEL, second_text, key)
         assert second.status == 429, second.body[:300]

tests/integration/streaming/test_text_completion_stream_guardrail_rebuild.py

@@ -16,7 +16,6 @@ from integration._support.client import (
     list_value,
     object_value,
 )
-from integration._support.database import read_rows
 from integration._support.process import OwnedProxy, WorkerHealthcheck, graceful_stop_seconds, owned_proxy_process
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import JsonValue
@@ -202,7 +201,7 @@ def test_guarded_text_completion_stream_reaches_the_caller_whole_with_its_usage(
     ], stream.frames
     assert _posted_targets(rig.wire) == ("/v1/completions",)
     (row,) = eventually(
-        lambda: read_rows(
+        lambda: rig.proxy.gateway.rows(
             'SELECT prompt_tokens, completion_tokens, metadata FROM "LiteLLM_SpendLogs" WHERE request_id = %s',
             (identity,),
         ),