mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-14 04:09:09 +00:00
test(integration): give every scripted responses reply its own id
The prompt-cache-breakpoint audit's scripted upstream minted the response id from the request's marker, so three identical marked requests shared one id. The spend-log writer skips rows whose request_id already landed, and the idempotent-logging cell saw one row for three calls on every leg. The upstream now mints a unique id per reply like a real provider, and the cells match a response to its request by the marker inside that id.
This commit is contained in:
parent
18a818530f
commit
0135dad700
3 changed files with 44 additions and 36 deletions
|
|
@ -29,6 +29,7 @@ INJECTION: Final[Mapping[str, JsonValue]] = {
|
|||
"prompt_cache_options": {"mode": "explicit"},
|
||||
}
|
||||
_SCRIPTED_FAILURE: Final = re.compile(r"fail-(\d{3})")
|
||||
_MINTED_RESPONSE: Final = re.compile(r"^resp_([0-9a-f]{32})-[0-9a-f]{32}$")
|
||||
_STARTED_WORKER: Final = re.compile(r"Started server process \[(\d+)\]")
|
||||
|
||||
Kind: TypeAlias = Literal["text", "image_url", "file", "input_audio"]
|
||||
|
|
@ -49,7 +50,7 @@ def _scripted(request: Request) -> Reply:
|
|||
if failure is not None:
|
||||
return rv.error(int(failure.group(1)), f"scripted {failure.group(1)} marker-{marker}", "scripted_failure")
|
||||
return responses_reply(
|
||||
f"resp_{marker or uuid.uuid4().hex}",
|
||||
f"resp_{marker or uuid.uuid4().hex}-{uuid.uuid4().hex}",
|
||||
string_value(body["model"]),
|
||||
rv.answer(marker),
|
||||
stream=body.get("stream") is True,
|
||||
|
|
@ -59,8 +60,15 @@ def _scripted(request: Request) -> Reply:
|
|||
respond: Final = answering_model_discovery(_scripted)
|
||||
|
||||
|
||||
def response_id(marker: str) -> str:
|
||||
return f"resp_{marker}"
|
||||
def response_marker(identity: str) -> str | None:
|
||||
minted: Final = tuple(
|
||||
found for candidate in rv.response_identities(identity) if (found := _MINTED_RESPONSE.match(candidate))
|
||||
)
|
||||
return minted[0].group(1) if minted else None
|
||||
|
||||
|
||||
def answers(identity: str, marker: str) -> bool:
|
||||
return response_marker(identity) == marker
|
||||
|
||||
|
||||
def prompt(marker: str) -> str:
|
||||
|
|
@ -161,7 +169,7 @@ class SpendLogs:
|
|||
return ROWS.validate_python(cursor.fetchall())
|
||||
|
||||
def landed(
|
||||
self, model: str, call_id: str, identity: str | None, *, status: str = "success", seconds: float = 70
|
||||
self, model: str, call_id: str, marker: str | None, *, status: str = "success", seconds: float = 70
|
||||
) -> dict[str, JsonValue]:
|
||||
rows: Final = eventually(
|
||||
lambda: self.rows_for(model),
|
||||
|
|
@ -172,7 +180,7 @@ class SpendLogs:
|
|||
assert len(matching) == 1, rows
|
||||
(row,) = matching
|
||||
assert row["status"] == status, row
|
||||
assert identity is None or rv.same_response(string_value(row["request_id"]), identity), (row, identity)
|
||||
assert marker is None or answers(string_value(row["request_id"]), marker), (row, marker)
|
||||
return row
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -112,7 +112,7 @@ def _chat(
|
|||
def _completion(response: httpx.Response, marker: str) -> str:
|
||||
assert response.status_code == 200, response.text
|
||||
body: Final = rv.JSON_OBJECT.validate_json(response.text)
|
||||
assert rv.same_response(string_value(body["id"]), pcb.response_id(marker)), body
|
||||
assert pcb.answers(string_value(body["id"]), marker), body
|
||||
(choice,) = rv.ITEMS.validate_python(body["choices"])
|
||||
assert rv.JSON_OBJECT.validate_python(choice["message"])["content"] == rv.answer(marker), body
|
||||
return response.headers["x-litellm-call-id"]
|
||||
|
|
@ -140,10 +140,10 @@ def test_openai_sdk_sends_a_valid_marker_through_the_bridge(bridge: _Bridge) ->
|
|||
model=bridge.on, messages=[_user(marker, pcb.EXPLICIT)], extra_body=dict(pcb.NO_CACHE)
|
||||
)
|
||||
completion: Final = raw.parse()
|
||||
assert rv.same_response(completion.id, pcb.response_id(marker)), completion
|
||||
assert pcb.answers(completion.id, marker), completion
|
||||
assert completion.choices[0].message.content == rv.answer(marker), completion
|
||||
pcb.assert_marker(_user_block_on_wire(bridge, marker), pcb.EXPLICIT)
|
||||
bridge.spend.landed(bridge.on, raw.headers["x-litellm-call-id"], pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.on, raw.headers["x-litellm-call-id"], marker)
|
||||
|
||||
|
||||
def test_openai_sdk_stream_carries_the_system_list_marker(bridge: _Bridge) -> None:
|
||||
|
|
@ -157,7 +157,7 @@ def test_openai_sdk_stream_carries_the_system_list_marker(bridge: _Bridge) -> No
|
|||
extra_body=dict(pcb.NO_CACHE),
|
||||
)
|
||||
chunks: Final = tuple(raw.parse())
|
||||
assert chunks and rv.same_response(chunks[0].id, pcb.response_id(marker)), chunks
|
||||
assert chunks and pcb.answers(chunks[0].id, marker), chunks
|
||||
streamed: Final = "".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices)
|
||||
assert streamed == rv.answer(marker), chunks
|
||||
request: Final = pcb.posted(bridge.wire, marker)
|
||||
|
|
@ -165,7 +165,7 @@ def test_openai_sdk_stream_carries_the_system_list_marker(bridge: _Bridge) -> No
|
|||
(system_block,) = pcb.content_of(pcb.input_items(request), "system")
|
||||
assert system_block["type"] == "input_text" and system_block["text"] == "sys", system_block
|
||||
pcb.assert_marker(system_block, pcb.EXPLICIT)
|
||||
bridge.spend.landed(bridge.on, raw.headers["x-litellm-call-id"], pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.on, raw.headers["x-litellm-call-id"], marker)
|
||||
|
||||
|
||||
async def test_async_openai_sdk_keeps_the_ttl_without_drop_params(bridge: _Bridge) -> None:
|
||||
|
|
@ -175,10 +175,10 @@ async def test_async_openai_sdk_keeps_the_ttl_without_drop_params(bridge: _Bridg
|
|||
model=bridge.off, messages=[_user(marker, pcb.EXPLICIT_30M)], extra_body=dict(pcb.NO_CACHE)
|
||||
)
|
||||
completion: Final = raw.parse()
|
||||
assert rv.same_response(completion.id, pcb.response_id(marker)), completion
|
||||
assert pcb.answers(completion.id, marker), completion
|
||||
assert completion.choices[0].message.content == rv.answer(marker), completion
|
||||
pcb.assert_marker(_user_block_on_wire(bridge, marker), pcb.EXPLICIT_30M)
|
||||
bridge.spend.landed(bridge.off, raw.headers["x-litellm-call-id"], pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.off, raw.headers["x-litellm-call-id"], marker)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ("on", "off"))
|
||||
|
|
@ -187,7 +187,7 @@ def test_valid_marker_shapes_reach_the_wire_unchanged(bridge: _Bridge, mode: Mod
|
|||
marker: Final = uuid.uuid4().hex
|
||||
call_id: Final = _completion(_chat(bridge, bridge.model(mode), [_user(marker, breakpoint)]), marker)
|
||||
pcb.assert_marker(_user_block_on_wire(bridge, marker), breakpoint)
|
||||
bridge.spend.landed(bridge.model(mode), call_id, pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.model(mode), call_id, marker)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -197,7 +197,7 @@ def test_marker_with_an_unknown_key(bridge: _Bridge, mode: Mode, expected: JsonV
|
|||
marker: Final = uuid.uuid4().hex
|
||||
call_id: Final = _completion(_chat(bridge, bridge.model(mode), [_user(marker, _UNKNOWN_KEY)]), marker)
|
||||
pcb.assert_marker(_user_block_on_wire(bridge, marker), expected)
|
||||
bridge.spend.landed(bridge.model(mode), call_id, pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.model(mode), call_id, marker)
|
||||
|
||||
|
||||
def _second_block_on_wire(bridge: _Bridge, marker: str) -> dict[str, JsonValue]:
|
||||
|
|
@ -220,7 +220,7 @@ def test_valid_marker_is_carried_on_every_block_kind(bridge: _Bridge, kind: pcb.
|
|||
second: Final = _second_block_on_wire(bridge, marker)
|
||||
assert second["type"] == pcb.WIRE_TYPE[kind], second
|
||||
pcb.assert_marker(second, pcb.EXPLICIT)
|
||||
bridge.spend.landed(bridge.model(mode), call_id, pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.model(mode), call_id, marker)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("kind", pcb.KINDS)
|
||||
|
|
@ -231,7 +231,7 @@ def test_malformed_marker_is_dropped_on_every_block_kind(bridge: _Bridge, kind:
|
|||
second: Final = _second_block_on_wire(bridge, marker)
|
||||
assert second["type"] == pcb.WIRE_TYPE[kind], second
|
||||
pcb.assert_marker(second, None)
|
||||
bridge.spend.landed(bridge.on, call_id, pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.on, call_id, marker)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("mode", "breakpoint", "expected"), _CASE_VALUES, ids=_CASE_IDS)
|
||||
|
|
@ -252,7 +252,7 @@ def test_tool_output_marker(bridge: _Bridge, mode: Mode, breakpoint: JsonValue,
|
|||
(output,) = pcb.function_output(pcb.input_items(request), "call_1")
|
||||
assert output["type"] == "input_text" and output["text"] == "found it", output
|
||||
pcb.assert_marker(output, expected)
|
||||
bridge.spend.landed(bridge.model(mode), call_id, pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.model(mode), call_id, marker)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("mode", "breakpoint", "expected"), _CASE_VALUES, ids=_CASE_IDS)
|
||||
|
|
@ -269,7 +269,7 @@ def test_assistant_list_marker(bridge: _Bridge, mode: Mode, breakpoint: JsonValu
|
|||
(earlier,) = pcb.content_of(pcb.input_items(request), "assistant")
|
||||
assert earlier["type"] == "output_text" and earlier["text"] == "earlier answer", earlier
|
||||
pcb.assert_marker(earlier, expected)
|
||||
bridge.spend.landed(bridge.model(mode), call_id, pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.model(mode), call_id, marker)
|
||||
|
||||
|
||||
def test_injected_system_marker_survives_a_trailing_audio_block(bridge: _Bridge) -> None:
|
||||
|
|
@ -285,7 +285,7 @@ def test_injected_system_marker_survives_a_trailing_audio_block(bridge: _Bridge)
|
|||
assert audio["type"] == "input_text", audio
|
||||
assert string_value(audio["text"]).startswith("{'type': 'input_audio'"), audio
|
||||
pcb.assert_marker(audio, pcb.EXPLICIT)
|
||||
bridge.spend.landed(bridge.injecting_on, call_id, pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.injecting_on, call_id, marker)
|
||||
|
||||
|
||||
def test_injected_marker_on_a_string_system_message(bridge: _Bridge) -> None:
|
||||
|
|
@ -300,7 +300,7 @@ def test_injected_marker_on_a_string_system_message(bridge: _Bridge) -> None:
|
|||
assert body["prompt_cache_options"] == {"mode": "explicit"}, body
|
||||
system_block: Final = pcb.single_block(pcb.input_items(request), "system")
|
||||
assert system_block == {"type": "input_text", "text": "Answer briefly", "prompt_cache_breakpoint": pcb.EXPLICIT}
|
||||
bridge.spend.landed(bridge.injecting_off, call_id, pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.injecting_off, call_id, marker)
|
||||
|
||||
|
||||
def _anthropic(bridge: _Bridge) -> anthropic.Anthropic:
|
||||
|
|
@ -362,13 +362,13 @@ def test_native_responses_request_never_enters_the_bridge(bridge: _Bridge) -> No
|
|||
)
|
||||
assert response.status_code == 200, response.text
|
||||
body: Final = rv.JSON_OBJECT.validate_json(response.text)
|
||||
assert rv.same_response(string_value(body["id"]), pcb.response_id(marker)), body
|
||||
assert pcb.answers(string_value(body["id"]), marker), body
|
||||
assert rv.answer(marker) in response.text, response.text
|
||||
request: Final = pcb.posted(bridge.wire, marker)
|
||||
_wire_body(request)
|
||||
on_wire: Final = pcb.single_block(pcb.input_items(request), "user")
|
||||
assert on_wire == pcb.marked(block, _UNKNOWN_KEY), on_wire
|
||||
bridge.spend.landed(bridge.on, response.headers["x-litellm-call-id"], pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.on, response.headers["x-litellm-call-id"], marker)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("breakpoint", _MALFORMED_VALUES, ids=_MALFORMED_IDS)
|
||||
|
|
@ -376,7 +376,7 @@ def test_malformed_marker_is_dropped_under_drop_params(bridge: _Bridge, breakpoi
|
|||
marker: Final = uuid.uuid4().hex
|
||||
call_id: Final = _completion(_chat(bridge, bridge.on, [_user(marker, breakpoint)]), marker)
|
||||
pcb.assert_marker(_user_block_on_wire(bridge, marker), None)
|
||||
bridge.spend.landed(bridge.on, call_id, pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.on, call_id, marker)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("breakpoint", _MALFORMED_VALUES, ids=_MALFORMED_IDS)
|
||||
|
|
@ -384,7 +384,7 @@ def test_malformed_marker_passes_verbatim_without_drop_params(bridge: _Bridge, b
|
|||
marker: Final = uuid.uuid4().hex
|
||||
call_id: Final = _completion(_chat(bridge, bridge.off, [_user(marker, breakpoint)]), marker)
|
||||
pcb.assert_marker(_user_block_on_wire(bridge, marker), breakpoint)
|
||||
bridge.spend.landed(bridge.off, call_id, pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.off, call_id, marker)
|
||||
|
||||
|
||||
def test_two_marked_blocks_are_both_carried(bridge: _Bridge) -> None:
|
||||
|
|
@ -399,7 +399,7 @@ def test_two_marked_blocks_are_both_carried(bridge: _Bridge) -> None:
|
|||
first, second = pcb.content_of(pcb.input_items(request), "user")
|
||||
assert first == {"type": "input_text", "text": pcb.prompt(marker), "prompt_cache_breakpoint": pcb.EXPLICIT}
|
||||
assert second == {"type": "input_text", "text": "and more", "prompt_cache_breakpoint": pcb.EXPLICIT_30M}
|
||||
bridge.spend.landed(bridge.on, call_id, pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.on, call_id, marker)
|
||||
|
||||
|
||||
def test_wrong_key_is_refused_before_the_wire(bridge: _Bridge) -> None:
|
||||
|
|
@ -427,7 +427,7 @@ def test_upstream_error_reaches_the_caller_once(bridge: _Bridge, mode: Mode, sta
|
|||
follow_up: Final = uuid.uuid4().hex
|
||||
call_id: Final = _completion(_chat(bridge, bridge.model(mode), [_user(follow_up, pcb.EXPLICIT)]), follow_up)
|
||||
pcb.assert_marker(_user_block_on_wire(bridge, follow_up), pcb.EXPLICIT)
|
||||
bridge.spend.landed(bridge.model(mode), call_id, pcb.response_id(follow_up))
|
||||
bridge.spend.landed(bridge.model(mode), call_id, follow_up)
|
||||
|
||||
|
||||
def test_null_drop_params_on_the_deployment_means_off(bridge: _Bridge) -> None:
|
||||
|
|
@ -436,7 +436,7 @@ def test_null_drop_params_on_the_deployment_means_off(bridge: _Bridge) -> None:
|
|||
model: Final = scenario.model(model=pcb.MODEL, api_base=bridge.api_base, drop_params=None)
|
||||
call_id: Final = _completion(_chat(bridge, model, [_user(marker, "yes")]), marker)
|
||||
pcb.assert_marker(_user_block_on_wire(bridge, marker), "yes")
|
||||
bridge.spend.landed(model, call_id, pcb.response_id(marker))
|
||||
bridge.spend.landed(model, call_id, marker)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ("on", "off"))
|
||||
|
|
@ -446,7 +446,7 @@ def test_null_or_missing_marker_sends_a_plain_block(bridge: _Bridge, shape: str,
|
|||
block: Final = pcb.marked(pcb.text(pcb.prompt(marker)), None) if shape == "null" else pcb.text(pcb.prompt(marker))
|
||||
call_id: Final = _completion(_chat(bridge, bridge.model(mode), [{"role": "user", "content": [block]}]), marker)
|
||||
assert _user_block_on_wire(bridge, marker) == {"type": "input_text", "text": pcb.prompt(marker)}
|
||||
bridge.spend.landed(bridge.model(mode), call_id, pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.model(mode), call_id, marker)
|
||||
|
||||
|
||||
async def _send_marked(client: httpx.AsyncClient, key: str, model: str, marker: str) -> httpx.Response:
|
||||
|
|
@ -494,7 +494,7 @@ async def test_flipping_drop_params_mid_burst_keeps_every_marked_request_answere
|
|||
flipped: Final = eventually(lambda: _probe_marker(bridge, model), lambda seen: seen == "yes", seconds=70)
|
||||
assert flipped == "yes"
|
||||
for marker, response in zip(markers, responses, strict=True):
|
||||
bridge.spend.landed(model, response.headers["x-litellm-call-id"], pcb.response_id(marker))
|
||||
bridge.spend.landed(model, response.headers["x-litellm-call-id"], marker)
|
||||
|
||||
|
||||
def test_three_identical_marked_requests_are_each_sent_and_logged(bridge: _Bridge) -> None:
|
||||
|
|
@ -507,4 +507,4 @@ def test_three_identical_marked_requests_are_each_sent_and_logged(bridge: _Bridg
|
|||
for request in posts:
|
||||
pcb.assert_marker(pcb.single_block(pcb.input_items(request), "user"), pcb.EXPLICIT)
|
||||
for call_id in call_ids:
|
||||
bridge.spend.landed(bridge.on, call_id, pcb.response_id(marker))
|
||||
bridge.spend.landed(bridge.on, call_id, marker)
|
||||
|
|
|
|||
|
|
@ -180,7 +180,7 @@ def _assert_answered_in_its_own_shape(served: _Served) -> None:
|
|||
assert served.text.startswith("{") != served.call.stream, served.text
|
||||
assert ("response.completed" in served.text) == (served.call.stream and served.call.endpoint == "responses")
|
||||
shown: Final = _upstream_id_shown_to_caller(served)
|
||||
assert shown is None or rv.same_response(shown, pcb.response_id(served.call.marker)), served.text
|
||||
assert shown is None or pcb.answers(shown, served.call.marker), served.text
|
||||
|
||||
|
||||
def _marked_once(posts: Sequence[Request], calls: Sequence[_Call], expected: JsonValue) -> None:
|
||||
|
|
@ -229,7 +229,7 @@ def _chat(gateway: Gateway, model: str, marker: str, breakpoint: JsonValue) -> h
|
|||
def _completion(response: httpx.Response, marker: str) -> str:
|
||||
assert response.status_code == 200, response.text
|
||||
body: Final = rv.JSON_OBJECT.validate_json(response.text)
|
||||
assert rv.same_response(string_value(body["id"]), pcb.response_id(marker)), body
|
||||
assert pcb.answers(string_value(body["id"]), marker), body
|
||||
assert rv.answer(marker) in response.text, response.text
|
||||
return response.headers["x-litellm-call-id"]
|
||||
|
||||
|
|
@ -245,11 +245,11 @@ def test_global_drop_params_drops_a_malformed_marker(global_rig: _GlobalRig, mod
|
|||
marker: Final = uuid.uuid4().hex
|
||||
call_id: Final = _completion(_chat(global_rig.gateway, model, marker, "yes"), marker)
|
||||
pcb.assert_marker(_user_block_on_wire(global_rig.wire, marker), None)
|
||||
spend.landed(model, call_id, pcb.response_id(marker))
|
||||
spend.landed(model, call_id, marker)
|
||||
control: Final = uuid.uuid4().hex
|
||||
control_id: Final = _completion(_chat(global_rig.gateway, model, control, pcb.EXPLICIT), control)
|
||||
pcb.assert_marker(_user_block_on_wire(global_rig.wire, control), pcb.EXPLICIT)
|
||||
spend.landed(model, control_id, pcb.response_id(control))
|
||||
spend.landed(model, control_id, control)
|
||||
|
||||
|
||||
async def test_mixed_burst_carries_every_marker_once(gateway: Gateway, spend: pcb.SpendLogs) -> None:
|
||||
|
|
@ -364,7 +364,7 @@ async def test_worker_sigkill_mid_burst_leaves_the_sibling_answering(
|
|||
follow_up: Final = uuid.uuid4().hex
|
||||
call_id: Final = _completion(_chat(candidate, _GLOBAL_UNSET, follow_up, "yes"), follow_up)
|
||||
pcb.assert_marker(_user_block_on_wire(wire, follow_up), None)
|
||||
spend.landed(_GLOBAL_UNSET, call_id, pcb.response_id(follow_up))
|
||||
spend.landed(_GLOBAL_UNSET, call_id, follow_up)
|
||||
|
||||
|
||||
async def test_proxy_restart_mid_burst_never_lands_a_served_call_twice(
|
||||
|
|
@ -388,7 +388,7 @@ async def test_proxy_restart_mid_burst_never_lands_a_served_call_twice(
|
|||
follow_up: Final = uuid.uuid4().hex
|
||||
call_id: Final = _completion(_chat(second.gateway, _GLOBAL_UNSET, follow_up, pcb.EXPLICIT), follow_up)
|
||||
pcb.assert_marker(_user_block_on_wire(wire, follow_up), pcb.EXPLICIT)
|
||||
spend.landed(_GLOBAL_UNSET, call_id, pcb.response_id(follow_up))
|
||||
spend.landed(_GLOBAL_UNSET, call_id, follow_up)
|
||||
counts: Final = Counter(string_value(row["litellm_call_id"]) for row in spend.rows_for(_GLOBAL_UNSET))
|
||||
assert all(count == 1 for count in counts.values()), counts
|
||||
landed: Final = sum(1 for item in served if item.call_id in counts)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue