scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-31 06:56:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 245

test(responses): guard stale retries and malformed comparison envelopes

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
ryan 2026-10-31 04:22:08 +00:00
parent 4eefe343b1
commit b9833798be
2 changed files with 45 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
tests/integration/providers/test_responses_core_wire.py
View file

@ -17,8 +17,8 @@ from openai.types.responses import ResponseCompletedEvent
from pydantic import JsonValue, TypeAdapter
import litellm
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.llms.openai import ResponseCompletedEvent as LiteLLMResponseCompletedEvent
from litellm.types.llms.openai import ResponsesAPIResponse
_MODEL: Final = "gpt-5"
_API_KEY: Final = "synthetic-responses-key"
@ -169,6 +169,9 @@ async def test_regression_sdk_comparison_response_id_is_raw_on_the_wire(
id="foreign-provider-wrapper",
),
pytest.param("resp_" + base64.b64encode(b"custom;id;response").decode(), id="unmanaged-semicolons"),
pytest.param(
"resp_" + base64.b64encode(b"openai;foreign;resp_unmanaged").decode(), id="openai-without-envelope-markers"
),
),
)
def test_guard_sdk_custom_host_preserves_unmanaged_comparison_ids(extra: bool, comparison_id: str) -> None:

41
tests/unit/test_responses_id_security.py
View file

@ -910,6 +910,47 @@ async def test_regression_comparison_id_reentry_keeps_ownership_after_owner_retr
assert _comparison_options(retried, nested) == {"comparison_response_id": _FABRICATED_PROVIDER_RESPONSE_ID}
@pytest.mark.asyncio
@pytest.mark.parametrize("nested", (False, True))
@pytest.mark.parametrize("replacement", ("resp_vendor_raw", "resp_!!!not-base64", ""))
async def test_guard_comparison_id_replaced_after_owner_hook_reaches_provider_unchanged(
salt_key_env: str, nested: bool, replacement: str
) -> None:
hook: Final = ResponsesIDSecurity(general_settings_reader=lambda: {}, signing_key_reader=lambda: salt_key_env)
owner: Final = UserAPIKeyAuth(user_id="owner-user", team_id="owner-team")
cache: Final = DualCache()
issued: Final = ResponsesAPIResponse.model_validate(
await hook.async_post_call_success_hook( # pyright: ignore[reportUnknownMemberType] # Legacy dict boundary
data={},
user_api_key_dict=owner,
response=ResponsesAPIResponse(
id=_FABRICATED_PROVIDER_RESPONSE_ID, created_at=1234567890, output=[], status="completed"
),
)
)
first_options: Final = {"comparison_response_id": issued.id}
first_request: Final = (
{"extra_body": {"prompt_cache_options": first_options}} if nested else {"prompt_cache_options": first_options}
)
first: Final = TypeAdapter(dict[str, object]).validate_python(
await hook.async_pre_call_hook( # pyright: ignore[reportUnknownMemberType] # Legacy dict boundary
user_api_key_dict=owner, cache=cache, data=first_request, call_type="aresponses"
)
)
replaced_options: Final = {"comparison_response_id": replacement, "mode": "implicit"}
replaced: Final = (
{**first, "extra_body": {"prompt_cache_options": replaced_options}}
if nested
else {**first, "prompt_cache_options": replaced_options}
)
rerun: Final = TypeAdapter(dict[str, object]).validate_python(
await hook.async_pre_call_hook( # pyright: ignore[reportUnknownMemberType] # Legacy dict boundary
user_api_key_dict=owner, cache=cache, data=replaced, call_type="aresponses"
)
)
assert _comparison_options(rerun, nested) == replaced_options, rerun
class TestUnrecognizedResponseIdIsRejected:
"""An id this proxy never issued carries no owner, so it must not reach the provider."""