scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-16 04:29:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 138

test(integration): cover responses endpoint contracts (#44733)

Browse source 
* test(integration): cover responses endpoint contracts

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): tighten responses contract tests per review

Isolate team-only response-id ownership with a same-user different-team key,
send every denial through the OpenAI SDK on all three aliases, make the
previous_response_id affinity check deterministic, assert the whole cancel
response, prove the Cursor stream text, -fast suffixes and include_obfuscation
strip, add SDK compact cases, restore router pre-call checks after the
affinity test, and inline the root-level responses_test_support helpers

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): wait for every proxy worker before routing to new deployments

Restore the cross-worker convergence wait dropped when the shared helper was
inlined, so the affinity and retrieve-routing tests cannot pass before the
alternative deployment is loaded on a multi-worker gateway

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): cover Responses ownership and wire contracts

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): address responses contract review

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): expect the provider model in the conversation passthrough body

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): expand responses endpoint contracts

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): restore streamed response retrieval

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): wait for proxy workers after deployment setup

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(integration): track readiness per proxy worker

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: mateo <mateo@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-10-14 06:59:15 +00:00 • committed by GitHub
parent 8822e086ec
commit 5261a2cc06
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 2357 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

64
tests/integration/_support/worker_readiness.py Normal file
View file

@ -0,0 +1,64 @@
from __future__ import annotations
import time
from collections.abc import Iterator, Mapping
from itertools import accumulate
from types import MappingProxyType
from typing import Final
import httpx
from integration._support.client import Gateway, JSON_OBJECT, object_value
from litellm.constants import PROXY_CONFIG_RELOAD_INTERVAL_SECONDS
def _record_latest_readback(
latest_by_worker: Mapping[int, int],
observation: tuple[int, int],
) -> Mapping[int, int]:
worker_pid, count = observation
return MappingProxyType({**latest_by_worker, worker_pid: count})
def wait_for_worker_readiness(
gateway: Gateway,
group: str,
deployments: int,
workers: int,
) -> None:
deadline: Final = time.monotonic() + (30 if workers == 1 else PROXY_CONFIG_RELOAD_INTERVAL_SECONDS * 2 + 30)
headers: Final = {"Authorization": f"Bearer {gateway.key}"}
def readbacks() -> Iterator[tuple[int, int]]:
while time.monotonic() < deadline:
timeout: Final = deadline - time.monotonic()
with httpx.Client(
base_url=str(gateway.client.base_url),
timeout=timeout,
trust_env=False,
verify=False,
) as client:
debug_response: Final = client.get("/debug/memory/summary", headers=headers)
response: Final = client.get(
"/model/info",
headers=headers,
)
assert debug_response.status_code == 200, debug_response.text
assert response.status_code == 200, response.text
assert debug_response.extensions.get("network_stream") is not None
assert debug_response.extensions.get("network_stream") is response.extensions.get("network_stream")
debug_body: Final = JSON_OBJECT.validate_json(debug_response.content)
worker_pid: Final = debug_body.get("worker_pid")
assert isinstance(worker_pid, int), debug_body
body: Final = JSON_OBJECT.validate_json(response.content)
entries: Final = body.get("data")
assert isinstance(entries, list), body
count: Final = sum(object_value(entry).get("model_name") == group for entry in entries)
yield worker_pid, count
initial_counts: Final[Mapping[int, int]] = MappingProxyType({})
latest_counts: Final = accumulate(readbacks(), _record_latest_readback, initial=initial_counts)
for latest_by_worker in latest_counts:
if len(latest_by_worker) >= workers and all(count == deployments for count in latest_by_worker.values()):
return
raise AssertionError(f"Expected {deployments} deployments on {workers} workers before the readiness deadline")

487
tests/integration/authorization/test_responses_id_ownership.py Normal file
View file

@ -0,0 +1,487 @@
from __future__ import annotations
import json
import os
import uuid
from collections.abc import Callable
from pathlib import Path
from typing import Final, Literal
from urllib.parse import urlsplit
import httpx
import openai
import pytest
from integration._support.client import Gateway, Scenario, object_value
from integration._support.process import owned_proxy
from integration._support.responses_vendor import same_response
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.worker_readiness import wait_for_worker_readiness
from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter
_MODEL: Final = "gpt-5"
_API_KEY: Final = "synthetic-responses-key"
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
_PROXY_WORKERS: Final = max(1, int(os.environ.get("INTEGRATION_PROXY_WORKERS", "1")))
_ALIASES: Final = ("/v1", "", "/openai/v1")
_DISABLE_HINT: Final = (
"To disable this security feature, set general_settings::disable_responses_id_security to True "
"in the config.yaml file."
)
_OTHER_USER_DETAIL: Final = (
f"Forbidden. The response id is not associated with the user, who this key belongs to. {_DISABLE_HINT}"
)
_OTHER_TEAM_DETAIL: Final = (
f"Forbidden. The response id is not associated with the team, who this key belongs to. {_DISABLE_HINT}"
)
_UNMANAGED_DETAIL: Final = (
"Forbidden. This response id was not issued by this proxy, so the proxy cannot tell who owns it. "
"To let keys address responses this proxy did not issue, set "
"general_settings::allow_unmanaged_response_ids to True in the config.yaml file."
)
class _DeleteResponse(BaseModel):
model_config = ConfigDict(extra="forbid")
id: str
object: Literal["response"]
deleted: bool
def _deployments_in_group(gateway: Gateway, group: str) -> int:
entries: Final = gateway.get("/model/info")["data"]
assert isinstance(entries, list), entries
return sum(object_value(entry).get("model_name") == group for entry in entries)
def _wait_for_workers(gateway: Gateway, group: str, deployments: int) -> None:
wait_for_worker_readiness(gateway, group, deployments, _PROXY_WORKERS)
def _model_discovery_reply() -> Reply:
return Reply(
body=json.dumps(
{"object": "list", "data": [{"id": _MODEL, "object": "model", "created": 1, "owned_by": "openai"}]}
).encode()
)
def _contract_requests(wire: Wire) -> tuple[Request, ...]:
requests: Final = wire.drain()
discovery: Final = tuple(request for request in requests if request.target == "/v1/models")
assert all(request.method == "GET" and request.body == b"" for request in discovery), requests
return tuple(request for request in requests if request.target != "/v1/models")
def _response(identity: str, status: str = "completed") -> bytes:
return json.dumps(
{
"id": identity,
"object": "response",
"created_at": 1,
"status": status,
"model": _MODEL,
"output": [],
"parallel_tool_calls": False,
"tool_choice": "auto",
"tools": [],
}
).encode()
def _deployment(scenario: Scenario, wire: Wire) -> str:
model: Final = scenario.model(
model=f"openai/{_MODEL}",
api_key=_API_KEY,
api_base=f"{wire.url}/v1",
)
_wait_for_workers(scenario.gateway, model, 1)
return model
def _key(scenario: Scenario, team_id: str, user_id: str, model: str) -> str:
return scenario.key(team_id=team_id, user_id=user_id, models=[model])
def _client(gateway: Gateway, key: str, prefix: str) -> openai.OpenAI:
return openai.OpenAI(base_url=f"{gateway.client.base_url}{prefix}", api_key=key, max_retries=0)
def _error(response: httpx.Response, detail: str) -> None:
assert response.status_code == 403, response.text
assert response.json() == {
"error": {
"message": detail,
"type": "permission_error",
"param": None,
"code": "403",
}
}, response.text
def _refused(call: Callable[[], object], detail: str) -> None:
with pytest.raises(openai.PermissionDeniedError) as refusal:
call()
_error(refusal.value.response, detail)
def _every_id_route(client: openai.OpenAI, model: str, response_id: str) -> tuple[Callable[[], object], ...]:
return (
lambda: client.responses.retrieve(response_id),
lambda: client.responses.delete(response_id),
lambda: client.responses.cancel(response_id),
lambda: client.responses.input_items.list(response_id),
lambda: client.responses.create(model=model, input="cross-tenant follow-up", previous_response_id=response_id),
)
@pytest.mark.parametrize("prefix", _ALIASES)
def test_response_ids_are_scoped_to_the_issuing_user_and_team(gateway: Gateway, prefix: str) -> None:
raw_id: Final = f"resp_owned_{uuid.uuid4().hex}"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return _model_discovery_reply()
body: Final = _JSON_OBJECT.validate_json(request.body) if request.body else {}
if request.method == "POST" and request.target == "/v1/responses":
if body.get("input") == "create owned response":
return Reply(body=_response(raw_id))
return Reply(body=_response(raw_id))
if request.method == "GET" and urlsplit(request.target).path.endswith("/input_items"):
return Reply(body=b'{"data":[],"has_more":false,"object":"list"}')
if request.method == "GET":
return Reply(body=_response(raw_id))
if request.method == "DELETE":
return Reply(body=json.dumps({"id": raw_id, "object": "response", "deleted": True}).encode())
if request.method == "POST" and request.target.endswith("/cancel"):
return Reply(body=_response(raw_id, "cancelled"))
raise AssertionError(f"Unexpected upstream request: {request.method} {request.target}")
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
team_a: Final = scenario.team(models=[model])
team_b: Final = scenario.team(models=[model])
user_a: Final = scenario.member(team_a)
user_b: Final = scenario.member(team_b)
gateway.post("/team/member_add", {"team_id": team_b, "member": {"role": "user", "user_id": user_a}})
client_a: Final = _client(gateway, _key(scenario, team_a, user_a, model), prefix)
other_user_client: Final = _client(gateway, _key(scenario, team_b, user_b, model), prefix)
other_team_client: Final = _client(gateway, _key(scenario, team_b, user_a, model), prefix)
created: Final = client_a.responses.create(model=model, input="create owned response")
client_id: Final = created.id
assert client_id.startswith("resp_") and client_id != raw_id, created.model_dump_json()
created_requests: Final = _contract_requests(wire)
assert [(request.method, request.target, request.body) for request in created_requests] == [
("POST", "/v1/responses", created_requests[0].body)
], created_requests
assert _JSON_OBJECT.validate_json(created_requests[0].body) == {
"model": _MODEL,
"input": "create owned response",
}, created_requests[0].body
for call in _every_id_route(other_user_client, model, client_id):
_refused(call, _OTHER_USER_DETAIL)
for call in _every_id_route(other_team_client, model, client_id):
_refused(call, _OTHER_TEAM_DETAIL)
assert _contract_requests(wire) == ()
retrieved_response: Final = client_a.responses.with_raw_response.retrieve(client_id)
retrieved: Final = retrieved_response.parse()
assert retrieved.id != raw_id and same_response(retrieved.id, client_id), retrieved_response.text
assert retrieved.model_dump(mode="json", exclude_none=True) == {
"id": retrieved.id,
"object": "response",
"created_at": 1.0,
"status": "completed",
"model": _MODEL,
"output": [],
"parallel_tool_calls": False,
"tool_choice": "auto",
"tools": [],
}, retrieved_response.text
deleted: Final = client_a.responses.with_raw_response.delete(client_id)
assert deleted.status_code == 200, deleted.text
deleted_response: Final = _DeleteResponse.model_validate_json(deleted.content)
assert deleted_response.model_dump(exclude={"id"}) == {"object": "response", "deleted": True}, deleted.text
assert same_response(deleted_response.id, client_id), deleted.text
cancelled_response: Final = client_a.responses.with_raw_response.cancel(client_id)
cancelled: Final = cancelled_response.parse()
assert cancelled.status == "cancelled", cancelled_response.text
assert cancelled.id != raw_id and same_response(cancelled.id, client_id), cancelled_response.text
assert cancelled.model_dump(mode="json", exclude_none=True) == {
"id": cancelled.id,
"object": "response",
"created_at": 1.0,
"status": "cancelled",
"model": _MODEL,
"output": [],
"parallel_tool_calls": False,
"tool_choice": "auto",
"tools": [],
}, cancelled_response.text
page_response: Final = client_a.responses.input_items.with_raw_response.list(client_id)
page: Final = page_response.parse()
assert page.model_dump(mode="json", exclude_none=True) == {
"data": [],
"has_more": False,
"object": "list",
}, page_response.text
followed: Final = client_a.responses.create(
model=model,
input="owned follow-up",
previous_response_id=client_id,
)
assert followed.id != raw_id and same_response(followed.id, client_id), followed.model_dump_json()
requests: Final = _contract_requests(wire)
assert [(request.method, urlsplit(request.target).path, request.body) for request in requests] == [
("GET", f"/v1/responses/{raw_id}", b""),
("DELETE", f"/v1/responses/{raw_id}", b""),
("POST", f"/v1/responses/{raw_id}/cancel", b"{}"),
("GET", f"/v1/responses/{raw_id}/input_items", b""),
("POST", "/v1/responses", requests[-1].body),
], requests
assert _JSON_OBJECT.validate_json(requests[-1].body) == {
"input": "owned follow-up",
"model": _MODEL,
"previous_response_id": raw_id,
}, requests[-1].body
unmanaged_id: Final = f"resp_vendor_{uuid.uuid4().hex}"
for call in _every_id_route(client_a, model, unmanaged_id):
_refused(call, _UNMANAGED_DETAIL)
assert _contract_requests(wire) == ()
@pytest.mark.parametrize("prefix", _ALIASES)
def test_compact_previous_response_id_is_scoped_to_the_issuing_user_and_team(
gateway: Gateway,
prefix: str,
) -> None:
pytest.skip(
"BUG: POST /v1/responses/compact does not check previous_response_id ownership; "
"another team's id and unmanaged ids reach the provider"
)
raw_id: Final = f"resp_compact_owned_{uuid.uuid4().hex}"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return _model_discovery_reply()
if request.method == "POST" and request.target == "/v1/responses":
return Reply(body=_response(raw_id))
return Reply(body=_response(raw_id))
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
team_a: Final = scenario.team(models=[model])
team_b: Final = scenario.team(models=[model])
user_a: Final = scenario.member(team_a)
user_b: Final = scenario.member(team_b)
gateway.post("/team/member_add", {"team_id": team_b, "member": {"role": "user", "user_id": user_a}})
client_a: Final = _client(gateway, _key(scenario, team_a, user_a, model), prefix)
other_user_client: Final = _client(gateway, _key(scenario, team_b, user_b, model), prefix)
other_team_client: Final = _client(gateway, _key(scenario, team_b, user_a, model), prefix)
created: Final = client_a.responses.create(model=model, input="create compact response")
client_id: Final = created.id
assert client_id != raw_id, created.model_dump_json()
assert len(_contract_requests(wire)) == 1
_refused(
lambda: other_user_client.responses.compact(
model=model,
previous_response_id=client_id,
input=[
{"role": "user", "content": [{"type": "input_text", "text": "compact someone else's response"}]}
],
),
_OTHER_USER_DETAIL,
)
_refused(
lambda: other_team_client.responses.compact(
model=model,
previous_response_id=client_id,
input=[
{"role": "user", "content": [{"type": "input_text", "text": "compact someone else's response"}]}
],
),
_OTHER_TEAM_DETAIL,
)
unmanaged_id: Final = f"resp_vendor_{uuid.uuid4().hex}"
_refused(
lambda: client_a.responses.compact(
model=model,
previous_response_id=unmanaged_id,
input=[
{"role": "user", "content": [{"type": "input_text", "text": "compact someone else's response"}]}
],
),
_UNMANAGED_DETAIL,
)
assert _contract_requests(wire) == ()
def test_allow_unmanaged_response_ids_forward_the_vendor_id_on_create(
gateway: Gateway,
tmp_path: Path,
) -> None:
raw_id: Final = f"resp_vendor_{uuid.uuid4().hex}"
raw_followup_id: Final = f"resp_followup_{uuid.uuid4().hex}"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return _model_discovery_reply()
return Reply(body=_response(raw_followup_id))
config: Final = tmp_path / "allow_unmanaged_responses.yaml"
config.write_text("general_settings:\n allow_unmanaged_response_ids: true\n")
with wire_server(respond) as wire, wire_server(respond) as decoy, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
team: Final = scenario.team(models=[model])
user: Final = scenario.member(team)
key: Final = _key(scenario, team, user, model)
with owned_proxy(
gateway,
tmp_path,
{
"INTEGRATION_PROXY_WORKERS": "1",
"OPENAI_BASE_URL": f"{decoy.url}/v1",
"OPENAI_API_KEY": "synthetic-env-fallback-key",
},
config=config,
) as allowed_gateway:
client: Final = _client(allowed_gateway, key, "/v1")
followed: Final = client.responses.create(
model=model,
input="allowed vendor follow-up",
previous_response_id=raw_id,
)
assert followed.model_dump(mode="json", exclude_none=True) == {
"id": followed.id,
"object": "response",
"created_at": 1.0,
"status": "completed",
"model": model,
"output": [],
"parallel_tool_calls": False,
"tool_choice": "auto",
"tools": [],
}, followed.model_dump_json()
retrieved_response: Final = client.responses.with_raw_response.retrieve(followed.id)
retrieved: Final = retrieved_response.parse()
assert same_response(retrieved.id, followed.id), retrieved_response.text
assert retrieved.model_dump(mode="json", exclude_none=True) == {
"id": retrieved.id,
"object": "response",
"created_at": 1.0,
"status": "completed",
"model": _MODEL,
"output": [],
"parallel_tool_calls": False,
"tool_choice": "auto",
"tools": [],
}, retrieved_response.text
requests: Final = _contract_requests(wire)
assert decoy.drain() == (), "The environment fallback deployment received a request"
assert [(request.method, request.target, request.body) for request in requests] == [
("POST", "/v1/responses", requests[0].body),
("GET", f"/v1/responses/{raw_followup_id}", b""),
], requests
assert all(request.headers["authorization"] == f"Bearer {_API_KEY}" for request in requests), requests
assert _JSON_OBJECT.validate_json(requests[0].body) == {
"model": _MODEL,
"input": "allowed vendor follow-up",
"previous_response_id": raw_id,
}, requests[0].body
@pytest.mark.parametrize("route", ("retrieve", "delete", "cancel", "input_items"))
def test_allow_unmanaged_response_id_lifecycle_route_reaches_provider(
gateway: Gateway,
tmp_path: Path,
route: Literal["retrieve", "delete", "cancel", "input_items"],
) -> None:
pytest.skip(
"BUG: unmanaged response-id lifecycle routes select model=None and return 400 instead of reaching OpenAI"
)
raw_id: Final = f"resp_vendor_{uuid.uuid4().hex}"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return _model_discovery_reply()
if request.method == "GET" and request.target == f"/v1/responses/{raw_id}/input_items":
return Reply(body=b'{"data":[],"has_more":false,"object":"list"}')
if request.method == "DELETE":
return Reply(body=json.dumps({"id": raw_id, "object": "response", "deleted": True}).encode())
if request.method == "POST" and request.target.endswith("/cancel"):
return Reply(body=_response(raw_id, "cancelled"))
return Reply(body=_response(raw_id))
config: Final = tmp_path / "allow_unmanaged_responses.yaml"
config.write_text("general_settings:\n allow_unmanaged_response_ids: true\n")
with wire_server(respond) as wire, wire_server(respond) as decoy, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
team: Final = scenario.team(models=[model])
user: Final = scenario.member(team)
key: Final = _key(scenario, team, user, model)
with owned_proxy(
gateway,
tmp_path,
{
"INTEGRATION_PROXY_WORKERS": "1",
"OPENAI_BASE_URL": f"{decoy.url}/v1",
"OPENAI_API_KEY": "synthetic-env-fallback-key",
},
config=config,
) as allowed_gateway:
client: Final = _client(allowed_gateway, key, "/v1")
expected_request: Final = {
"retrieve": ("GET", f"/v1/responses/{raw_id}", b""),
"delete": ("DELETE", f"/v1/responses/{raw_id}", b""),
"cancel": ("POST", f"/v1/responses/{raw_id}/cancel", b"{}"),
"input_items": ("GET", f"/v1/responses/{raw_id}/input_items", b""),
}[route]
match route:
case "retrieve":
retrieved: Final = client.responses.retrieve(raw_id)
assert retrieved.model_dump(mode="json", exclude_none=True) == {
"id": raw_id,
"object": "response",
"created_at": 1.0,
"status": "completed",
"model": _MODEL,
"output": [],
"parallel_tool_calls": False,
"tool_choice": "auto",
"tools": [],
}, retrieved.model_dump_json()
case "delete":
deleted: Final = client.responses.delete(raw_id)
assert deleted.model_dump(mode="json", exclude_none=True) == {
"id": raw_id,
"object": "response",
"deleted": True,
}, deleted.model_dump_json()
case "cancel":
cancelled: Final = client.responses.cancel(raw_id)
assert cancelled.model_dump(mode="json", exclude_none=True) == {
"id": raw_id,
"object": "response",
"created_at": 1.0,
"status": "cancelled",
"model": _MODEL,
"output": [],
"parallel_tool_calls": False,
"tool_choice": "auto",
"tools": [],
}, cancelled.model_dump_json()
case "input_items":
page: Final = client.responses.input_items.list(raw_id)
assert page.model_dump(mode="json", exclude_none=True) == {
"data": [],
"has_more": False,
"object": "list",
}, page.model_dump_json()
requests: Final = _contract_requests(wire)
assert tuple((request.method, request.target, request.body) for request in requests) == (expected_request,), (
requests
)
assert all(request.headers["authorization"] == f"Bearer {_API_KEY}" for request in requests), requests
assert decoy.drain() == (), "The environment fallback deployment received a request"

450
tests/integration/providers/test_cursor_responses_wire.py Normal file
View file

@ -0,0 +1,450 @@
from __future__ import annotations
import json
import os
import uuid
from typing import Final
import pytest
from integration._support.client import Gateway, Scenario
from integration._support.responses_vendor import same_response
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.worker_readiness import wait_for_worker_readiness
from pydantic import JsonValue, TypeAdapter
_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")
_RESPONSES_SUFFIXES: Final[tuple[tuple[str, dict[str, JsonValue]], ...]] = tuple(
(f"-thinking-{level}", {"reasoning": {"effort": level}}) for level in _THINKING_LEVELS
) + (
("-fast", {}),
("-thinking-low-fast", {"reasoning": {"effort": "low"}}),
("-thinking-high-fast", {"reasoning": {"effort": "high"}}),
)
_CHAT_SUFFIXES: Final[tuple[tuple[str, str], ...]] = (("", ""),) + tuple(
(f"-thinking-{level}", level) for level in _CHAT_THINKING_LEVELS
)
_PROMPT: Final = "Inspect the requested change."
_INPUT: Final = [{"role": "user", "content": [{"type": "input_text", "text": _PROMPT}]}]
_CUSTOM_TOOL: Final = {
"type": "custom",
"name": "ApplyPatch",
"description": "Apply a patch",
"format": {"type": "grammar", "syntax": "lark", "definition": 'start: "ok"'},
}
_FUNCTION_TOOL: Final = {
"type": "function",
"name": "lookup",
"description": "Look up a record",
"parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]},
"strict": True,
}
def model_discovery_reply(model: str) -> Reply:
return Reply(
body=json.dumps(
{"object": "list", "data": [{"id": model, "object": "model", "created": 1, "owned_by": "openai"}]}
).encode()
)
def drain_contract_requests(wire: Wire) -> tuple[Request, ...]:
requests: Final = wire.drain()
discovery: Final = tuple(request for request in requests if request.target == "/v1/models")
assert all(request.method == "GET" and request.body == b"" for request in discovery), requests
return tuple(request for request in requests if request.target != "/v1/models")
def _response(identity: str) -> dict[str, JsonValue]:
return {
"id": identity,
"object": "response",
"created_at": 1,
"status": "completed",
"model": _MODEL,
"output": [
{
"id": "msg_cursor",
"type": "message",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "Patch inspected.", "annotations": []}],
}
],
"usage": {"input_tokens": 4, "output_tokens": 3, "total_tokens": 7},
}
def _stream(identity: str) -> tuple[bytes, ...]:
response: Final = _response(identity)
return tuple(
f"data: {json.dumps(event)}\n\n".encode()
for event in (
{"type": "response.created", "response": {**response, "status": "in_progress", "output": []}},
{
"type": "response.output_text.delta",
"item_id": "msg_cursor",
"output_index": 0,
"content_index": 0,
"delta": "Patch inspected.",
},
{"type": "response.completed", "response": response},
)
) + (b"data: [DONE]\n\n",)
def _chat_completion(stream: bool) -> bytes | tuple[bytes, ...]:
if stream:
return (
b'data: {"id":"chat_cursor_1","object":"chat.completion.chunk","created":1,"model":"gpt-5","choices":[{"index":0,"delta":{"content":"Patch inspected."},"finish_reason":null}]}\n\n',
b'data: {"id":"chat_cursor_1","object":"chat.completion.chunk","created":1,"model":"gpt-5","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\n',
b"data: [DONE]\n\n",
)
return json.dumps(
{
"id": "chat_cursor_1",
"object": "chat.completion",
"created": 1,
"model": _MODEL,
"system_fingerprint": "fp_cursor",
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": "Patch inspected."},
"finish_reason": "stop",
}
],
"usage": {"prompt_tokens": 4, "completion_tokens": 3, "total_tokens": 7},
}
).encode()
def _deployment(scenario: Scenario, wire: Wire) -> str:
model: Final = scenario.model(
model=f"openai/{_MODEL}",
api_key=_API_KEY,
api_base=f"{wire.url}/v1",
)
wait_for_worker_readiness(scenario.gateway, model, 1, _PROXY_WORKERS)
return model
@pytest.mark.parametrize(
("suffix", "reasoning"),
_RESPONSES_SUFFIXES,
)
@pytest.mark.parametrize("stream", (False, True))
def test_cursor_responses_input_uses_responses_wire_and_chat_output(
gateway: Gateway,
stream: bool,
suffix: str,
reasoning: dict[str, JsonValue],
) -> None:
raw_id: Final = f"resp_cursor_{uuid.uuid4().hex}"
with gateway.scenario() as scenario:
team_alias: Final = f"integration-{uuid.uuid4().hex}"
team: Final = scenario.team(team_alias=team_alias)
user: Final = scenario.member(team)
key: Final = scenario.key(team_id=team, user_id=user)
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
assert request.method == "POST"
assert request.headers["authorization"] == f"Bearer {_API_KEY}", dict(request.headers)
if request.target == "/v1/chat/completions":
completion: Final = _chat_completion(stream)
return Reply(
content_type="text/event-stream" if stream else "application/json",
chunks=completion if isinstance(completion, tuple) else None,
body=completion if isinstance(completion, bytes) else b"",
)
assert request.target == "/v1/responses", request.target
body: Final = _JSON_OBJECT.validate_json(request.body)
assert {key: value for key, value in body.items() if key != "metadata"} == {
"model": _MODEL,
"input": _INPUT,
**reasoning,
"tools": [_CUSTOM_TOOL],
"stream": stream,
}, request.body
if stream:
return Reply(content_type="text/event-stream", chunks=_stream(raw_id))
return Reply(body=json.dumps(_response(raw_id)).encode())
with wire_server(respond) as wire:
model: Final = _deployment(scenario, wire)
response: Final = gateway.request(
"POST",
"/cursor/chat/completions",
{
"model": f"{model}{suffix}",
"input": _INPUT,
"messages": [],
"tools": [_CUSTOM_TOOL],
"stream_options": {"include_usage": True, "include_obfuscation": False},
"stream": stream,
},
key=key,
)
assert response.status_code == 200, response.text
if stream:
events: Final = tuple(
_JSON_OBJECT.validate_json(line.removeprefix("data: "))
for line in response.text.splitlines()
if line.startswith("data: ") and line != "data: [DONE]"
)
assert events, response.text
assert all(event.get("object") == "chat.completion.chunk" for event in events), response.text
assert all(not str(event.get("type", "")).startswith("response.") for event in events), response.text
assert [{k: v for k, v in event.items() if k not in ("id", "created")} for event in events] == [
{
"object": "chat.completion.chunk",
"model": model,
"choices": [{"index": 0, "delta": {"role": "assistant", "content": "Patch inspected."}}],
},
{
"object": "chat.completion.chunk",
"model": model,
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
},
], response.text
else:
payload: Final = _JSON_OBJECT.validate_json(response.content)
identity: Final = payload.get("id")
created: Final = payload.get("created")
assert isinstance(identity, str) and same_response(identity, raw_id), response.text
assert isinstance(created, int) and created > 0, response.text
assert payload == {
"id": identity,
"created": created,
"model": model,
"object": "chat.completion",
"system_fingerprint": None,
"choices": [
{
"finish_reason": "stop",
"index": 0,
"message": {
"content": "Patch inspected.",
"role": "assistant",
"tool_calls": None,
"function_call": None,
"provider_specific_fields": None,
},
}
],
"usage": {
"completion_tokens": 3,
"prompt_tokens": 4,
"total_tokens": 7,
"completion_tokens_details": None,
"prompt_tokens_details": None,
},
}, response.text
assert [request.target for request in drain_contract_requests(wire)] == ["/v1/responses"]
def test_cursor_responses_body_carries_no_proxy_metadata(gateway: Gateway) -> None:
pytest.skip(
"BUG: /cursor/chat/completions forwards proxy-internal metadata (key hash, team, user, client ip, user agent) "
"to the provider in the Responses body"
)
raw_id: Final = f"resp_cursor_metadata_{uuid.uuid4().hex}"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
assert request.method == "POST" and request.target == "/v1/responses", request.target
return Reply(body=json.dumps(_response(raw_id)).encode())
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
response: Final = gateway.request(
"POST",
"/cursor/chat/completions",
{
"model": f"{model}-thinking-high",
"input": _INPUT,
"messages": [],
"tools": [_CUSTOM_TOOL],
},
)
assert response.status_code == 200, response.text
requests: Final = drain_contract_requests(wire)
assert len(requests) == 1, requests
body: Final = _JSON_OBJECT.validate_json(requests[0].body)
assert body == {
"model": _MODEL,
"input": _INPUT,
"reasoning": {"effort": "high"},
"tools": [_CUSTOM_TOOL],
}, requests[0].body
def test_cursor_stream_reports_usage_when_the_client_asks_for_it(gateway: Gateway) -> None:
pytest.skip(
"BUG: /cursor/chat/completions with stream_options.include_usage streams no usage chunk; "
"the Responses usage from response.completed never reaches the chat client"
)
raw_id: Final = f"resp_cursor_usage_{uuid.uuid4().hex}"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
assert (request.method, request.target) == ("POST", "/v1/responses"), request.target
return Reply(content_type="text/event-stream", chunks=_stream(raw_id))
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
response: Final = gateway.request(
"POST",
"/cursor/chat/completions",
{
"model": model,
"input": _INPUT,
"stream_options": {"include_usage": True},
"stream": True,
},
)
assert response.status_code == 200, response.text
events: Final = tuple(
_JSON_OBJECT.validate_json(line.removeprefix("data: "))
for line in response.text.splitlines()
if line.startswith("data: ") and line != "data: [DONE]"
)
assert [event["usage"] for event in events if event.get("usage")] == [
{"completion_tokens": 3, "prompt_tokens": 4, "total_tokens": 7}
], response.text
assert [request.target for request in drain_contract_requests(wire)] == ["/v1/responses"]
@pytest.mark.parametrize(("suffix", "reasoning_effort"), _CHAT_SUFFIXES)
@pytest.mark.parametrize("stream", (False, True))
def test_cursor_messages_body_uses_chat_completions_wire(
gateway: Gateway,
stream: bool,
suffix: str,
reasoning_effort: str,
) -> None:
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
completion: Final = _chat_completion(stream)
return Reply(
content_type="text/event-stream" if stream else "application/json",
chunks=completion if isinstance(completion, tuple) else None,
body=completion if isinstance(completion, bytes) else b"",
)
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
team: Final = scenario.team(models=[model])
user: Final = scenario.member(team)
key: Final = scenario.key(team_id=team, user_id=user, models=[model])
response: Final = gateway.request(
"POST",
"/cursor/chat/completions",
{
"model": f"{model}{suffix}",
"messages": [{"role": "user", "content": "genuine chat request"}],
"tools": [_FUNCTION_TOOL, _CUSTOM_TOOL],
"tool_choice": {"type": "function", "name": "lookup"},
"stream": stream,
},
key=key,
)
assert response.status_code == 200, response.text
if stream:
events: Final = tuple(
_JSON_OBJECT.validate_json(line.removeprefix("data: "))
for line in response.text.splitlines()
if line.startswith("data: ") and line != "data: [DONE]"
)
assert [
{key: value for key, value in event.items() if key not in ("id", "created")} for event in events
] == [
{
"object": "chat.completion.chunk",
"model": model,
"choices": [{"index": 0, "delta": {"role": "assistant", "content": "Patch inspected."}}],
},
{
"object": "chat.completion.chunk",
"model": model,
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
},
], response.text
else:
payload: Final = _JSON_OBJECT.validate_json(response.content)
identity: Final = payload.get("id")
created: Final = payload.get("created")
assert isinstance(identity, str) and same_response(identity, "chat_cursor_1"), response.text
assert isinstance(created, int) and created > 0, response.text
assert payload == {
"id": identity,
"created": created,
"model": model,
"object": "chat.completion",
"system_fingerprint": "fp_cursor",
"choices": [
{
"finish_reason": "stop",
"index": 0,
"message": {
"content": "Patch inspected.",
"role": "assistant",
"provider_specific_fields": {"refusal": None},
},
"provider_specific_fields": {},
}
],
"usage": {
"completion_tokens": 3,
"prompt_tokens": 4,
"total_tokens": 7,
},
}, response.text
requests: Final = drain_contract_requests(wire)
assert [(request.method, request.target) for request in requests] == [("POST", "/v1/chat/completions")], (
requests
)
assert _JSON_OBJECT.validate_json(requests[0].body) == {
"model": _MODEL,
"messages": [{"role": "user", "content": "genuine chat request"}],
"tools": [
{
"type": "function",
"function": {
"name": "lookup",
"description": "Look up a record",
"parameters": {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
"strict": True,
},
},
{
"type": "custom",
"custom": {
"name": "ApplyPatch",
"description": "Apply a patch",
"format": {
"type": "grammar",
"grammar": {"syntax": "lark", "definition": 'start: "ok"'},
},
},
},
],
"tool_choice": {"type": "function", "function": {"name": "lookup"}},
**({"stream": True} if stream else {}),
**({"stream_options": {"include_usage": True}} if stream else {}),
**({"reasoning_effort": reasoning_effort} if reasoning_effort else {}),
}, requests[0].body

367
tests/integration/providers/test_responses_compact_wire.py Normal file
View file

@ -0,0 +1,367 @@
from __future__ import annotations
import json
import os
import uuid
from typing import Final
import openai
import pytest
from integration._support.client import Gateway, Scenario
from integration._support.responses_vendor import same_response
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.worker_readiness import wait_for_worker_readiness
from openai.types.responses import CompactedResponse
from pydantic import JsonValue, TypeAdapter
_MODEL: Final = "gpt-5"
_API_KEY: Final = "synthetic-responses-key"
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
_PROXY_WORKERS: Final = max(1, int(os.environ.get("INTEGRATION_PROXY_WORKERS", "1")))
_ALIASES: Final = ("/v1/responses/compact", "/responses/compact", "/openai/v1/responses/compact")
_FUNCTION_TOOL: Final = {
"type": "function",
"name": "lookup",
"description": "Look up a record",
"parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]},
"strict": True,
}
def model_discovery_reply(model: str) -> Reply:
return Reply(
body=json.dumps(
{"object": "list", "data": [{"id": model, "object": "model", "created": 1, "owned_by": "openai"}]}
).encode()
)
def drain_contract_requests(wire: Wire) -> tuple[Request, ...]:
requests: Final = wire.drain()
discovery: Final = tuple(request for request in requests if request.target == "/v1/models")
assert all(request.method == "GET" and request.body == b"" for request in discovery), requests
return tuple(request for request in requests if request.target != "/v1/models")
def _deployment(scenario: Scenario, wire: Wire) -> str:
model: Final = scenario.model(
model=f"openai/{_MODEL}",
api_key=_API_KEY,
api_base=f"{wire.url}/v1",
)
wait_for_worker_readiness(scenario.gateway, model, 1, _PROXY_WORKERS)
return model
def test_responses_compact_routes_forward_complete_request_and_typed_items(gateway: Gateway) -> None:
reasoning_item: Final = {
"id": f"rs_{uuid.uuid4().hex}",
"type": "reasoning",
"summary": [{"type": "summary_text", "text": "Prior reasoning"}],
"encrypted_content": "synthetic-encrypted-content",
}
function_output_item: Final = {
"type": "function_call_output",
"call_id": "call_previous",
"output": "The record is ready.",
}
requests: Final = (
{
"input": "Compact this conversation.",
},
{
"input": [reasoning_item, function_output_item],
"instructions": "Preserve the latest decision.",
"tools": [_FUNCTION_TOOL],
},
)
output_items: Final = [
{
"id": f"rs_compact_{uuid.uuid4().hex}",
"type": "reasoning",
"summary": [{"type": "summary_text", "text": "Compacted state"}],
"encrypted_content": "synthetic-compact-state",
},
{
"id": "fc_output_compacted",
"type": "function_call_output",
"call_id": "call_previous",
"output": "The record is ready.",
"status": "completed",
},
]
completed_id: Final = f"resp_compact_{uuid.uuid4().hex}"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
assert request.method == "POST"
assert request.headers["authorization"] == f"Bearer {_API_KEY}", dict(request.headers)
body: Final = _JSON_OBJECT.validate_json(request.body)
compact_params: Final = (
{
"model": _MODEL,
**requests[0],
}
if body.get("input") == requests[0]["input"]
else {
"model": _MODEL,
**requests[1],
}
)
assert body == compact_params, request.body
return Reply(
body=json.dumps(
{
"id": completed_id,
"created_at": 1,
"object": "response.compaction",
"output": output_items,
"usage": {
"input_tokens": 7,
"input_tokens_details": {"cached_tokens": 0},
"output_tokens": 3,
"output_tokens_details": {"reasoning_tokens": 0},
"total_tokens": 10,
},
}
).encode()
)
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
for alias in _ALIASES:
for request_body in requests:
response: Final = gateway.request(
"POST",
alias,
{"model": model, **request_body},
)
assert response.status_code == 200, response.text
compacted: Final = CompactedResponse.model_validate_json(response.content)
assert compacted.id != completed_id, compacted.model_dump_json()
assert same_response(compacted.id, completed_id), compacted.model_dump_json()
assert compacted.model_dump(mode="json") == {
"id": compacted.id,
"created_at": 1,
"error": None,
"incomplete_details": None,
"instructions": None,
"metadata": None,
"model": model,
"object": "response.compaction",
"output": [
{
"id": output_items[0]["id"],
"summary": [{"text": "Compacted state", "type": "summary_text"}],
"type": "reasoning",
"content": None,
"encrypted_content": "synthetic-compact-state",
"status": None,
},
{
"id": "fc_output_compacted",
"call_id": "call_previous",
"output": "The record is ready.",
"status": "completed",
"type": "function_call_output",
"created_by": None,
},
],
"parallel_tool_calls": None,
"temperature": None,
"tool_choice": None,
"tools": None,
"top_p": None,
"max_output_tokens": None,
"previous_response_id": None,
"reasoning": None,
"status": None,
"text": None,
"truncation": None,
"usage": {
"input_tokens": 7,
"input_tokens_details": {
"audio_tokens": None,
"cached_tokens": 0,
"cached_tokens_details": None,
"image_tokens": None,
"text_tokens": None,
"video_tokens": None,
},
"output_tokens": 3,
"output_tokens_details": {
"audio_tokens": None,
"reasoning_tokens": 0,
"text_tokens": None,
},
"total_tokens": 10,
"cost": None,
},
"user": None,
"store": None,
}, response.text
assert [request.target for request in drain_contract_requests(wire)] == ["/v1/responses/compact"] * 6
@pytest.mark.parametrize("prefix", ("/v1", "", "/openai/v1"))
def test_sdk_compact_reaches_the_compact_endpoint_with_its_full_body(gateway: Gateway, prefix: str) -> None:
completed_id: Final = f"resp_compact_sdk_{uuid.uuid4().hex}"
input_items: Final = [
{"role": "user", "content": [{"type": "input_text", "text": "Summarize the patch so far."}]},
{
"id": f"rs_{uuid.uuid4().hex}",
"type": "reasoning",
"summary": [{"type": "summary_text", "text": "Prior reasoning"}],
"encrypted_content": "synthetic-encrypted-content",
},
{"type": "function_call_output", "call_id": "call_previous", "output": "The record is ready."},
]
output_item: Final = {
"id": f"rs_compact_{uuid.uuid4().hex}",
"type": "reasoning",
"summary": [{"type": "summary_text", "text": "Compacted state"}],
"encrypted_content": "synthetic-compact-state",
}
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
assert (request.method, request.target) == ("POST", "/v1/responses/compact"), request.target
assert request.headers["authorization"] == f"Bearer {_API_KEY}", dict(request.headers)
assert _JSON_OBJECT.validate_json(request.body) == {
"model": _MODEL,
"input": input_items,
"instructions": "Preserve the latest decision.",
}, request.body
return Reply(
body=json.dumps(
{
"id": completed_id,
"created_at": 1,
"object": "response.compaction",
"output": [output_item],
"usage": {
"input_tokens": 7,
"input_tokens_details": {"cached_tokens": 0},
"output_tokens": 3,
"output_tokens_details": {"reasoning_tokens": 0},
"total_tokens": 10,
},
}
).encode()
)
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
client: Final = openai.OpenAI(base_url=f"{gateway.client.base_url}{prefix}", api_key=gateway.key, max_retries=0)
compacted: Final = client.responses.compact(
model=model,
input=input_items,
instructions="Preserve the latest decision.",
)
assert compacted.id != completed_id and same_response(compacted.id, completed_id), compacted.model_dump_json()
assert compacted.model_dump(mode="json", exclude_none=True) == {
"id": compacted.id,
"created_at": 1,
"model": model,
"object": "response.compaction",
"output": [output_item],
"usage": {
"input_tokens": 7,
"input_tokens_details": {"cached_tokens": 0},
"output_tokens": 3,
"output_tokens_details": {"reasoning_tokens": 0},
"total_tokens": 10,
},
}, compacted.model_dump_json()
assert [(request.method, request.target) for request in drain_contract_requests(wire)] == [
("POST", "/v1/responses/compact")
]
@pytest.mark.parametrize("prefix", ("/v1", "", "/openai/v1"))
def test_compact_previous_response_id_reaches_the_provider_as_the_raw_id(
gateway: Gateway,
prefix: str,
) -> None:
pytest.skip("BUG: compact forwards the proxy-encrypted previous_response_id instead of the raw provider id")
raw_id: Final = f"resp_compact_prev_{uuid.uuid4().hex}"
completed_id: Final = f"resp_compact_result_{uuid.uuid4().hex}"
input_items: Final = [{"role": "user", "content": [{"type": "input_text", "text": "Summarize the patch so far."}]}]
output_item: Final = {
"id": f"rs_compact_{uuid.uuid4().hex}",
"type": "reasoning",
"summary": [{"type": "summary_text", "text": "Compacted patch"}],
"encrypted_content": "synthetic-compact-state",
}
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
if request.method == "POST" and request.target == "/v1/responses":
return Reply(
body=json.dumps(
{
"id": raw_id,
"object": "response",
"created_at": 1,
"status": "completed",
"model": _MODEL,
"output": [],
}
).encode()
)
return Reply(
body=json.dumps(
{
"id": completed_id,
"created_at": 1,
"object": "response.compaction",
"output": [output_item],
"usage": {
"input_tokens": 7,
"input_tokens_details": {"cached_tokens": 0},
"output_tokens": 3,
"output_tokens_details": {"reasoning_tokens": 0},
"total_tokens": 10,
},
}
).encode()
)
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
client: Final = openai.OpenAI(
base_url=f"{gateway.client.base_url}{prefix}",
api_key=gateway.key,
max_retries=0,
)
created: Final = client.responses.create(model=model, input="create before compact")
assert created.id != raw_id, created.model_dump_json()
compact_response: Final = client.responses.with_raw_response.compact(
model=model,
previous_response_id=created.id,
input=input_items,
)
compacted: Final = compact_response.parse()
assert compacted.object == "response.compaction", compact_response.http_response.text
assert compacted.created_at == 1, compact_response.http_response.text
assert same_response(compacted.id, completed_id), compact_response.http_response.text
assert len(compacted.output) == 1, compact_response.http_response.text
assert compacted.output[0].type == "reasoning", compact_response.http_response.text
assert compacted.output[0].summary[0].text == "Compacted patch", compact_response.http_response.text
requests: Final = drain_contract_requests(wire)
assert [(request.method, request.target) for request in requests] == [
("POST", "/v1/responses"),
("POST", "/v1/responses/compact"),
], requests
assert _JSON_OBJECT.validate_json(requests[0].body) == {
"model": _MODEL,
"input": "create before compact",
}, requests[0].body
assert _JSON_OBJECT.validate_json(requests[1].body) == {
"model": _MODEL,
"previous_response_id": raw_id,
"input": input_items,
}, compact_response.http_response.text

406
tests/integration/providers/test_responses_core_wire.py Normal file
View file

@ -0,0 +1,406 @@
from __future__ import annotations
import json
import os
import uuid
from typing import Final
import openai
import pytest
from integration._support.client import Gateway, Scenario, object_value, string_value
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.worker_readiness import wait_for_worker_readiness
from openai.types.responses import ResponseCompletedEvent
from pydantic import JsonValue, TypeAdapter
_MODEL: Final = "gpt-5"
_API_KEY: Final = "synthetic-responses-key"
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
_PROXY_WORKERS: Final = max(1, int(os.environ.get("INTEGRATION_PROXY_WORKERS", "1")))
_ALIASES: Final = ("/v1", "", "/openai/v1")
def _deployments_in_group(gateway: Gateway, group: str) -> int:
entries: Final = gateway.get("/model/info")["data"]
assert isinstance(entries, list), entries
return sum(object_value(entry).get("model_name") == group for entry in entries)
def _wait_for_workers(gateway: Gateway, group: str, deployments: int) -> None:
wait_for_worker_readiness(gateway, group, deployments, _PROXY_WORKERS)
def model_discovery_reply(model: str) -> Reply:
return Reply(
body=json.dumps(
{"object": "list", "data": [{"id": model, "object": "model", "created": 1, "owned_by": "openai"}]}
).encode()
)
def drain_contract_requests(wire: Wire) -> tuple[Request, ...]:
requests: Final = wire.drain()
discovery: Final = tuple(request for request in requests if request.target == "/v1/models")
assert all(request.method == "GET" and request.body == b"" for request in discovery), requests
return tuple(request for request in requests if request.target != "/v1/models")
def _response(identity: str, status: str = "completed") -> bytes:
return json.dumps(
{
"id": identity,
"object": "response",
"created_at": 1,
"status": status,
"model": _MODEL,
"output": [],
}
).encode()
def _stream(identity: str) -> tuple[bytes, ...]:
return tuple(
f"data: {json.dumps(event)}\n\n".encode()
for event in (
{"type": "response.created", "response": json.loads(_response(identity, "in_progress"))},
{"type": "response.completed", "response": json.loads(_response(identity))},
)
) + (b"data: [DONE]\n\n",)
def _register_in_group(scenario: Scenario, wire: Wire, group: str, order: int | None = None) -> str:
existing: Final = _deployments_in_group(scenario.gateway, group)
created: Final = scenario.gateway.post(
"/model/new",
{
"model_name": group,
"litellm_params": {
"model": f"openai/{_MODEL}",
"api_key": _API_KEY,
"api_base": f"{wire.url}/v1",
**({} if order is None else {"order": order}),
},
"model_info": {},
},
)
model_id: Final = string_value(object_value(created["model_info"])["id"])
scenario.cleanups.callback(scenario.delete_model, model_id)
_wait_for_workers(scenario.gateway, group, existing + 1)
return group
def _register(scenario: Scenario, wire: Wire, group: str | None = None) -> str:
if group is not None:
return _register_in_group(scenario, wire, group)
model: Final = scenario.model(
model=f"openai/{_MODEL}",
api_key=_API_KEY,
api_base=f"{wire.url}/v1",
)
_wait_for_workers(scenario.gateway, model, 1)
return model
def _enable_responses_affinity(scenario: Scenario) -> None:
gateway: Final = scenario.gateway
settings: Final = object_value(gateway.get("/router/settings")["current_values"])
current: Final = settings.get("optional_pre_call_checks")
original: Final = list(current) if isinstance(current, list) else []
scenario.cleanups.callback(
gateway.post,
"/config/update",
{"router_settings": {"optional_pre_call_checks": original}},
)
gateway.post(
"/config/update",
{"router_settings": {"optional_pre_call_checks": [*original, "responses_api_deployment_check"]}},
)
@pytest.mark.parametrize("prefix", _ALIASES)
@pytest.mark.parametrize("stream", (False, True))
def test_previous_response_id_uses_the_raw_id_and_affinity(
gateway: Gateway,
prefix: str,
stream: bool,
) -> None:
first_raw_id: Final = f"resp_turn_one_{uuid.uuid4().hex}"
second_raw_id: Final = f"resp_turn_two_{uuid.uuid4().hex}"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
assert request.method == "POST"
assert request.target == "/v1/responses"
assert request.headers["authorization"] == f"Bearer {_API_KEY}", dict(request.headers)
body: Final = _JSON_OBJECT.validate_json(request.body)
if body["input"] == "turn one":
assert body == {"model": _MODEL, "input": "turn one"}, request.body
return Reply(body=_response(first_raw_id))
if stream:
return Reply(content_type="text/event-stream", chunks=_stream(second_raw_id))
return Reply(body=_response(second_raw_id))
def unused(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
assert request.method == "POST"
assert request.target == "/v1/responses"
assert request.headers["authorization"] == f"Bearer {_API_KEY}", dict(request.headers)
if stream:
return Reply(content_type="text/event-stream", chunks=_stream(second_raw_id))
return Reply(body=_response(second_raw_id))
with wire_server(respond) as first_wire, wire_server(unused) as second_wire, gateway.scenario() as scenario:
model: Final = f"integration-{uuid.uuid4().hex}"
_enable_responses_affinity(scenario)
assert _register(scenario, first_wire, model) == model
client: Final = openai.OpenAI(
base_url=f"{gateway.client.base_url}{prefix}",
api_key=gateway.key,
max_retries=0,
)
first: Final = client.responses.create(model=model, input="turn one")
assert first.id.startswith("resp_") and first.id != first_raw_id, first.model_dump_json()
_register_in_group(scenario, second_wire, model, order=1)
second_result: Final = client.responses.create(
model=model,
input="turn two",
previous_response_id=first.id,
stream=stream,
)
if stream:
events: Final = list(second_result)
completed: Final = events[-1]
assert isinstance(completed, ResponseCompletedEvent), events
assert completed.response.id != second_raw_id, completed.model_dump_json()
assert completed.response.output == [], completed.model_dump_json()
else:
assert second_result.id != second_raw_id, second_result.model_dump_json()
assert second_result.output == [], second_result.model_dump_json()
assert drain_contract_requests(second_wire) == (), "previous_response_id did not preserve deployment affinity"
requests: Final = drain_contract_requests(first_wire)
assert len(requests) == 2, requests
assert [_JSON_OBJECT.validate_json(request.body) for request in requests] == [
{"model": _MODEL, "input": "turn one"},
{
"model": _MODEL,
"input": "turn two",
"previous_response_id": first_raw_id,
"stream": stream,
},
], [request.body for request in requests]
@pytest.mark.parametrize("prefix", _ALIASES)
def test_codex_responses_request_preserves_all_supported_fields(gateway: Gateway, prefix: str) -> None:
response_id: Final = f"resp_codex_{uuid.uuid4().hex}"
input_items: Final = [
{
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "Inspect the requested change."}],
},
{
"type": "reasoning",
"id": f"rs_{uuid.uuid4().hex}",
"summary": [{"type": "summary_text", "text": "Plan the inspection"}],
"encrypted_content": f"gAAAA-codex-encrypted-{uuid.uuid4().hex}",
},
{
"type": "function_call",
"call_id": "call_lookup",
"name": "lookup",
"arguments": '{"query": "diff"}',
},
{
"type": "function_call_output",
"call_id": "call_lookup",
"output": '{"rows": 1}',
},
{
"type": "custom_tool_call",
"call_id": "call_patch",
"name": "ApplyPatch",
"input": "*** Begin Patch\n*** End Patch",
},
{
"type": "custom_tool_call_output",
"call_id": "call_patch",
"output": "Done",
},
]
function_tool: Final = {
"type": "function",
"name": "lookup",
"description": "Look up a record",
"parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]},
"strict": True,
}
custom_tool: Final = {
"type": "custom",
"name": "ApplyPatch",
"description": "Apply a patch",
"format": {"type": "grammar", "syntax": "lark", "definition": 'start: "ok"'},
}
web_search_tool: Final = {"type": "web_search"}
expected_body: Final = {
"model": _MODEL,
"input": input_items,
"stream": True,
"include": ["reasoning.encrypted_content"],
"store": False,
"prompt_cache_key": "cache-codex-contract",
"reasoning": {"effort": "high", "summary": "auto"},
"text": {
"verbosity": "low",
"format": {
"type": "json_schema",
"name": "patch_summary",
"strict": True,
"schema": {
"type": "object",
"properties": {"summary": {"type": "string"}},
"required": ["summary"],
"additionalProperties": False,
},
},
},
"tools": [function_tool, custom_tool, web_search_tool],
"tool_choice": "auto",
"parallel_tool_calls": True,
"instructions": "Return concise structured output.",
}
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
assert request.method == "POST"
assert request.target == "/v1/responses"
assert request.headers["authorization"] == f"Bearer {_API_KEY}", dict(request.headers)
_JSON_OBJECT.validate_json(request.body)
return Reply(content_type="text/event-stream", chunks=_stream(response_id))
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _register(scenario, wire)
client: Final = openai.OpenAI(
base_url=f"{gateway.client.base_url}{prefix}",
api_key=gateway.key,
max_retries=0,
)
stream: Final = client.responses.create(
model=model,
input=input_items,
stream=True,
include=["reasoning.encrypted_content"],
store=False,
prompt_cache_key="cache-codex-contract",
reasoning={"effort": "high", "summary": "auto"},
text={
"verbosity": "low",
"format": {
"type": "json_schema",
"name": "patch_summary",
"strict": True,
"schema": {
"type": "object",
"properties": {"summary": {"type": "string"}},
"required": ["summary"],
"additionalProperties": False,
},
},
},
tools=[function_tool, custom_tool, web_search_tool],
tool_choice="auto",
parallel_tool_calls=True,
instructions="Return concise structured output.",
)
events: Final = list(stream)
completed: Final = events[-1]
assert isinstance(completed, ResponseCompletedEvent), events
assert completed.response.status == "completed", completed.model_dump_json()
assert completed.response.output == [], completed.model_dump_json()
requests: Final = drain_contract_requests(wire)
assert len(requests) == 1, requests
assert _JSON_OBJECT.validate_json(requests[0].body) == expected_body, requests[0].body
assert [request.target for request in requests] == ["/v1/responses"]
@pytest.mark.parametrize("prefix", _ALIASES)
@pytest.mark.parametrize("case", ("mixed", "many"))
def test_responses_metadata_reaches_the_provider_as_typed(
gateway: Gateway,
prefix: str,
case: str,
) -> None:
metadata: Final = (
{
"string_value": "keep this",
"integer_value": 7,
"boolean_value": True,
"nested_value": {"key": "drop this"},
"second_string": "keep this too",
}
if case == "mixed"
else {f"key_{index:02d}": f"value_{index:02d}" for index in range(17)}
)
expected_metadata: Final = dict(
tuple((key, value) for key, value in metadata.items() if isinstance(value, str))[:16]
)
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
assert (request.method, request.target) == ("POST", "/v1/responses"), request.target
return Reply(body=_response(f"resp_metadata_{uuid.uuid4().hex}"))
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _register(scenario, wire)
response: Final = gateway.request(
"POST",
f"{prefix}/responses",
{
"model": model,
"input": "metadata propagation",
"metadata": metadata,
},
)
assert response.status_code == 200, response.text
requests: Final = drain_contract_requests(wire)
assert len(requests) == 1, requests
assert _JSON_OBJECT.validate_json(requests[0].body) == {
"model": _MODEL,
"input": "metadata propagation",
"metadata": expected_metadata,
}, requests[0].body
def test_conversation_is_forwarded_as_a_separate_responses_parameter(gateway: Gateway) -> None:
pytest.skip("BUG: POST /v1/responses drops the conversation param; upstream receives the body without it")
conversation: Final = {"id": f"conv_{uuid.uuid4().hex}"}
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
return Reply(body=_response(f"resp_conversation_{uuid.uuid4().hex}"))
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _register(scenario, wire)
client: Final = openai.OpenAI(
base_url=f"{gateway.client.base_url}/v1",
api_key=gateway.key,
max_retries=0,
)
response: Final = client.responses.create(
model=model,
input="continue this conversation",
conversation=conversation,
)
assert response.object == "response", response.model_dump_json()
requests: Final = drain_contract_requests(wire)
assert [(request.method, request.target) for request in requests] == [("POST", "/v1/responses")], requests
assert _JSON_OBJECT.validate_json(requests[0].body) == {
"model": _MODEL,
"input": "continue this conversation",
"conversation": conversation,
}, requests[0].body

583
tests/integration/providers/test_responses_lifecycle_wire.py Normal file
View file

@ -0,0 +1,583 @@
from __future__ import annotations
import json
import os
import uuid
from typing import Final, Literal
from urllib.parse import parse_qsl, urlsplit
import openai
import pytest
from integration._support.client import Gateway, Scenario, eventually, object_value, string_value
from integration._support.responses_vendor import same_response
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.worker_readiness import wait_for_worker_readiness
from openai.types.responses import Response, ResponseStreamEvent
from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter
_MODEL: Final = "gpt-5"
_API_KEY: Final = "synthetic-responses-key"
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
_PROXY_WORKERS: Final = max(1, int(os.environ.get("INTEGRATION_PROXY_WORKERS", "1")))
_ALIASES: Final = ("/v1", "", "/openai/v1")
class _DeleteResponse(BaseModel):
model_config = ConfigDict(extra="forbid")
id: str
object: Literal["response"]
deleted: bool
def _deployments_in_group(gateway: Gateway, group: str) -> int:
entries: Final = gateway.get("/model/info")["data"]
assert isinstance(entries, list), entries
return sum(object_value(entry).get("model_name") == group for entry in entries)
def _wait_for_workers(gateway: Gateway, group: str, deployments: int) -> None:
wait_for_worker_readiness(gateway, group, deployments, _PROXY_WORKERS)
def model_discovery_reply(model: str) -> Reply:
return Reply(
body=json.dumps(
{"object": "list", "data": [{"id": model, "object": "model", "created": 1, "owned_by": "openai"}]}
).encode()
)
def drain_contract_requests(wire: Wire) -> tuple[Request, ...]:
requests: Final = wire.drain()
discovery: Final = tuple(request for request in requests if request.target == "/v1/models")
assert all(request.method == "GET" and request.body == b"" for request in discovery), requests
return tuple(request for request in requests if request.target != "/v1/models")
def _retrieve_or_missing_deployment(client: openai.OpenAI, response_id: str) -> Response | openai.BadRequestError:
try:
return client.responses.retrieve(response_id)
except openai.BadRequestError as refusal:
if "no healthy deployments" not in str(refusal).lower():
raise
return refusal
def _response(identity: str, status: str = "completed") -> bytes:
return json.dumps(
{
"id": identity,
"object": "response",
"created_at": 1,
"status": status,
"model": _MODEL,
"output": [],
}
).encode()
def _stream(identity: str) -> tuple[bytes, ...]:
return tuple(
f"data: {json.dumps(event)}\n\n".encode()
for event in (
{"type": "response.created", "response": json.loads(_response(identity, "in_progress"))},
{"type": "response.completed", "response": json.loads(_response(identity))},
)
) + (b"data: [DONE]\n\n",)
def _deployment(scenario: Scenario, wire: Wire, api_key: str = _API_KEY) -> str:
model: Final = scenario.model(
model=f"openai/{_MODEL}",
api_key=api_key,
api_base=f"{wire.url}/v1",
)
_wait_for_workers(scenario.gateway, model, 1)
return model
def _stream_response_events(
client: openai.OpenAI, response_id: str
) -> tuple[tuple[ResponseStreamEvent, ...], str | None]:
try:
return (
tuple(
client.responses.retrieve(
response_id,
stream=True,
starting_after=3,
include=["reasoning.encrypted_content"],
)
),
None,
)
except openai.InternalServerError as error:
return (), error.response.text
def _register_preferred_in_group(scenario: Scenario, wire: Wire, group: str, api_key: str) -> None:
existing: Final = _deployments_in_group(scenario.gateway, group)
created: Final = scenario.gateway.post(
"/model/new",
{
"model_name": group,
"litellm_params": {
"model": f"openai/{_MODEL}",
"api_key": api_key,
"api_base": f"{wire.url}/v1",
"order": 1,
},
"model_info": {},
},
)
deployment_id: Final = string_value(object_value(created["model_info"])["id"])
scenario.cleanups.callback(scenario.delete_model, deployment_id)
_wait_for_workers(scenario.gateway, group, existing + 1)
def _client(gateway: Gateway, key: str, prefix: str) -> openai.OpenAI:
return openai.OpenAI(
base_url=f"{gateway.client.base_url}{prefix}",
api_key=key,
max_retries=0,
)
@pytest.mark.parametrize("prefix", _ALIASES)
def test_retrieve_routes_to_the_deployment_that_created_the_response(
gateway: Gateway,
prefix: str,
) -> None:
raw_id: Final = f"resp_affinity_{uuid.uuid4().hex}"
wire_b_key: Final = "synthetic-responses-key-b"
wire_a_key: Final = "synthetic-responses-key-a"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
if request.method == "POST":
return Reply(body=_response(raw_id))
return Reply(body=_response(raw_id))
with wire_server(respond) as wire_b, wire_server(respond) as wire_a, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire_b, wire_b_key)
client: Final = _client(gateway, gateway.key, prefix)
created: Final = client.responses.create(model=model, input="create for affinity")
client_id: Final = created.id
assert client_id.startswith("resp_") and client_id != raw_id, created.model_dump_json()
_register_preferred_in_group(scenario, wire_a, model, wire_a_key)
retrieved: Final = client.responses.retrieve(client_id)
assert retrieved.id != raw_id, retrieved.model_dump_json()
assert same_response(retrieved.id, client_id), retrieved.model_dump_json()
assert retrieved.model_dump(mode="json", exclude_none=True) == {
"id": retrieved.id,
"object": "response",
"created_at": 1.0,
"status": "completed",
"model": _MODEL,
"output": [],
}, retrieved.model_dump_json()
requests: Final = drain_contract_requests(wire_b)
assert [(request.method, request.target, request.body) for request in requests] == [
("POST", "/v1/responses", requests[0].body),
("GET", f"/v1/responses/{raw_id}", b""),
], requests
assert _JSON_OBJECT.validate_json(requests[0].body) == {
"model": _MODEL,
"input": "create for affinity",
}, requests[0].body
assert all(request.headers["authorization"] == f"Bearer {wire_b_key}" for request in requests), requests
assert requests[1].body == b"", requests[1]
assert drain_contract_requests(wire_a) == (), (
f"{prefix or '/responses'} did not preserve response deployment affinity"
)
@pytest.mark.parametrize("prefix", _ALIASES)
def test_retrieve_returns_the_id_the_client_holds(gateway: Gateway, prefix: str) -> None:
pytest.skip(
"BUG: GET /v1/responses/{id} returns a re-encrypted id that differs byte-for-byte from the id the client sent"
)
raw_id: Final = f"resp_retrieve_id_{uuid.uuid4().hex}"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
assert request.headers["authorization"] == f"Bearer {_API_KEY}", dict(request.headers)
if request.method == "POST":
assert request.target == "/v1/responses"
assert _JSON_OBJECT.validate_json(request.body) == {"model": _MODEL, "input": "retrieve client id"}, (
request.body
)
return Reply(body=_response(raw_id))
assert request.method == "GET"
assert request.target == f"/v1/responses/{raw_id}", request.target
assert request.body == b""
return Reply(body=_response(raw_id))
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
client: Final = _client(gateway, gateway.key, prefix)
created: Final = client.responses.create(model=model, input="retrieve client id")
client_id: Final = created.id
retrieved: Final = client.responses.retrieve(client_id)
requests: Final = drain_contract_requests(wire)
assert [(request.method, request.target, request.body) for request in requests] == [
("POST", "/v1/responses", requests[0].body),
("GET", f"/v1/responses/{raw_id}", b""),
], requests
assert _JSON_OBJECT.validate_json(requests[0].body) == {
"model": _MODEL,
"input": "retrieve client id",
}, requests[0].body
assert retrieved.id == client_id, (retrieved.model_dump_json(), requests)
@pytest.mark.parametrize("prefix", _ALIASES)
def test_streamed_response_ids_are_stable_and_retrievable_on_each_alias(
gateway: Gateway,
prefix: str,
) -> None:
raw_id: Final = f"resp_stream_{uuid.uuid4().hex}"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
assert request.headers["authorization"] == f"Bearer {_API_KEY}", dict(request.headers)
if request.method == "POST":
assert request.target == "/v1/responses"
assert _JSON_OBJECT.validate_json(request.body) == {
"model": _MODEL,
"input": "stream lifecycle",
"stream": True,
}, request.body
return Reply(content_type="text/event-stream", chunks=_stream(raw_id))
return Reply(body=_response(raw_id))
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
team: Final = scenario.team(models=[model])
user: Final = scenario.member(team)
key: Final = scenario.key(team_id=team, user_id=user, models=[model])
client: Final = _client(gateway, key, prefix)
stream: Final = client.responses.create(model=model, input="stream lifecycle", stream=True)
events: Final = list(stream)
response_ids: Final = tuple(event.response.id for event in events if hasattr(event, "response"))
assert response_ids and all(response_id == response_ids[0] for response_id in response_ids), events
streamed_id: Final = response_ids[0]
assert streamed_id.startswith("resp_") and streamed_id != raw_id, events
retrieved: Final = eventually(
lambda: _retrieve_or_missing_deployment(client, streamed_id),
lambda outcome: not isinstance(outcome, openai.BadRequestError),
seconds=60,
)
assert isinstance(retrieved, Response), retrieved
assert same_response(retrieved.id, streamed_id), retrieved.model_dump_json()
assert retrieved.model_dump(mode="json", exclude_none=True) == {
"id": retrieved.id,
"object": "response",
"created_at": 1.0,
"status": "completed",
"model": _MODEL,
"output": [],
}, retrieved.model_dump_json()
requests: Final = drain_contract_requests(wire)
assert [(request.method, urlsplit(request.target).path, request.body) for request in requests] == [
("POST", "/v1/responses", requests[0].body),
("GET", f"/v1/responses/{raw_id}", b""),
], requests
assert _JSON_OBJECT.validate_json(requests[0].body) == {
"model": _MODEL,
"input": "stream lifecycle",
"stream": True,
}, requests[0].body
assert all(request.headers["authorization"] == f"Bearer {_API_KEY}" for request in requests), requests
@pytest.mark.parametrize("prefix", _ALIASES)
def test_retrieve_query_reaches_the_provider(gateway: Gateway, prefix: str) -> None:
pytest.skip(
"BUG: GET /v1/responses/{id} drops include and starting_after; streamed GET returns 500 for provider SSE"
)
normal_raw_id: Final = f"resp_query_normal_{uuid.uuid4().hex}"
stream_raw_id: Final = f"resp_query_stream_{uuid.uuid4().hex}"
normal_include: Final = ["message.output_text.logprobs", "reasoning.encrypted_content"]
stream_include: Final = ["reasoning.encrypted_content"]
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
if request.method == "POST" and urlsplit(request.target).path == "/v1/responses":
body: Final = _JSON_OBJECT.validate_json(request.body)
identity: Final = stream_raw_id if body.get("input") == "query retrieve stream" else normal_raw_id
return Reply(body=_response(identity))
if urlsplit(request.target).path == f"/v1/responses/{stream_raw_id}":
return Reply(content_type="text/event-stream", chunks=_stream(stream_raw_id))
return Reply(body=_response(normal_raw_id))
with wire_server(respond) as wire:
sdk_client: Final = openai.OpenAI(base_url=f"{wire.url}/v1", api_key=_API_KEY, max_retries=0)
sdk_client.responses.retrieve(normal_raw_id, include=normal_include)
tuple(
sdk_client.responses.retrieve(
stream_raw_id,
stream=True,
starting_after=3,
include=stream_include,
)
)
sdk_requests: Final = drain_contract_requests(wire)
assert [request.method for request in sdk_requests] == ["GET", "GET"], sdk_requests
expected_queries: Final = tuple(
tuple(sorted(parse_qsl(urlsplit(request.target).query, keep_blank_values=True))) for request in sdk_requests
)
with gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
client: Final = _client(gateway, gateway.key, prefix)
normal_created: Final = client.responses.create(model=model, input="query retrieve normal")
stream_created: Final = client.responses.create(model=model, input="query retrieve stream")
retrieved_response: Final = client.responses.with_raw_response.retrieve(
normal_created.id,
include=normal_include,
)
retrieved: Final = retrieved_response.parse()
assert retrieved.id != normal_raw_id and same_response(retrieved.id, normal_created.id), (
retrieved_response.text
)
assert retrieved.model_dump(mode="json", exclude_none=True) == {
"id": retrieved.id,
"object": "response",
"created_at": 1.0,
"status": "completed",
"model": _MODEL,
"output": [],
}, retrieved_response.text
stream_result: Final = _stream_response_events(client, stream_created.id)
stream_events: Final = stream_result[0]
stream_error_text: Final = stream_result[1]
if stream_error_text is not None:
requests: Final = drain_contract_requests(wire)
assert [(request.method, urlsplit(request.target).path, request.body) for request in requests[2:]] == [
("GET", f"/v1/responses/{normal_raw_id}", b""),
("GET", f"/v1/responses/{stream_raw_id}", b""),
], stream_error_text
actual_queries: Final = tuple(
tuple(sorted(parse_qsl(urlsplit(request.target).query, keep_blank_values=True)))
for request in requests[2:]
)
assert actual_queries == expected_queries, (
requests[2:],
actual_queries,
expected_queries,
stream_error_text,
)
pytest.fail(f"Streaming retrieval returned an API error: {stream_error_text}")
completed_events: Final = tuple(event for event in stream_events if event.type == "response.completed")
assert len(completed_events) == 1, stream_events
completed: Final = completed_events[0]
assert completed.response.id != stream_raw_id and same_response(completed.response.id, stream_created.id), (
stream_events
)
assert completed.model_dump(mode="json", exclude_none=True) == {
"type": "response.completed",
"response": {
"id": completed.response.id,
"object": "response",
"created_at": 1.0,
"status": "completed",
"model": _MODEL,
"output": [],
},
}, stream_events
requests: Final = drain_contract_requests(wire)
assert [(request.method, urlsplit(request.target).path, request.body) for request in requests] == [
("POST", "/v1/responses", requests[0].body),
("POST", "/v1/responses", requests[1].body),
("GET", f"/v1/responses/{normal_raw_id}", b""),
("GET", f"/v1/responses/{stream_raw_id}", b""),
], requests
assert _JSON_OBJECT.validate_json(requests[0].body) == {
"model": _MODEL,
"input": "query retrieve normal",
}, requests[0].body
assert _JSON_OBJECT.validate_json(requests[1].body) == {
"model": _MODEL,
"input": "query retrieve stream",
}, requests[1].body
actual_queries: Final = tuple(
tuple(sorted(parse_qsl(urlsplit(request.target).query, keep_blank_values=True)))
for request in requests[2:]
)
assert actual_queries == expected_queries, requests
@pytest.mark.parametrize("prefix", _ALIASES)
def test_delete_forwards_the_provider_id_once_and_reports_deletion(
gateway: Gateway,
prefix: str,
) -> None:
raw_id: Final = f"resp_delete_{uuid.uuid4().hex}"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
assert request.headers["authorization"] == f"Bearer {_API_KEY}", dict(request.headers)
if request.method == "POST":
assert request.target == "/v1/responses"
assert _JSON_OBJECT.validate_json(request.body) == {"model": _MODEL, "input": "delete lifecycle"}, (
request.body
)
return Reply(body=_response(raw_id))
assert request.method == "DELETE"
assert request.body == b""
return Reply(body=json.dumps({"id": raw_id, "object": "response", "deleted": True}).encode())
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
client: Final = _client(gateway, gateway.key, prefix)
created: Final = client.responses.create(model=model, input="delete lifecycle")
deleted: Final = client.responses.with_raw_response.delete(created.id)
assert deleted.status_code == 200, deleted.text
parsed: Final = _DeleteResponse.model_validate_json(deleted.content)
assert same_response(parsed.id, created.id), deleted.text
assert parsed.model_dump(mode="json", exclude={"id"}) == {
"object": "response",
"deleted": True,
}, deleted.text
requests: Final = drain_contract_requests(wire)
assert [(request.method, request.target, request.body) for request in requests] == [
("POST", "/v1/responses", requests[0].body),
("DELETE", f"/v1/responses/{raw_id}", b""),
], requests
@pytest.mark.parametrize("prefix", _ALIASES)
def test_delete_returns_the_client_held_id_not_the_provider_id(
gateway: Gateway,
prefix: str,
) -> None:
pytest.skip("BUG: DELETE /v1/responses/{id} returns the raw provider id instead of the proxy-issued id")
raw_id: Final = f"resp_delete_id_{uuid.uuid4().hex}"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
assert request.headers["authorization"] == f"Bearer {_API_KEY}", dict(request.headers)
if request.method == "POST":
assert request.target == "/v1/responses"
assert _JSON_OBJECT.validate_json(request.body) == {"model": _MODEL, "input": "delete client id"}, (
request.body
)
return Reply(body=_response(raw_id))
assert request.method == "DELETE"
assert request.target == f"/v1/responses/{raw_id}", request.target
assert request.body == b""
return Reply(body=json.dumps({"id": raw_id, "object": "response", "deleted": True}).encode())
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
client: Final = _client(gateway, gateway.key, prefix)
created: Final = client.responses.create(model=model, input="delete client id")
deleted: Final = client.responses.with_raw_response.delete(created.id)
assert deleted.status_code == 200, deleted.text
parsed: Final = _DeleteResponse.model_validate_json(deleted.content)
requests: Final = drain_contract_requests(wire)
assert [(request.method, request.target, request.body) for request in requests] == [
("POST", "/v1/responses", requests[0].body),
("DELETE", f"/v1/responses/{raw_id}", b""),
], requests
assert _JSON_OBJECT.validate_json(requests[0].body) == {
"model": _MODEL,
"input": "delete client id",
}, requests[0].body
assert parsed.id != raw_id, (deleted.text, requests)
assert parsed.model_dump(mode="json") == {
"id": created.id,
"object": "response",
"deleted": True,
}, deleted.text
@pytest.mark.parametrize("prefix", _ALIASES)
def test_cancel_forwards_provider_id_and_returns_cancelled_response(
gateway: Gateway,
prefix: str,
) -> None:
raw_id: Final = f"resp_cancel_{uuid.uuid4().hex}"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
assert request.headers["authorization"] == f"Bearer {_API_KEY}", dict(request.headers)
if request.method == "POST" and request.target == "/v1/responses":
assert _JSON_OBJECT.validate_json(request.body) == {
"model": _MODEL,
"input": "cancel lifecycle",
"background": True,
}, request.body
return Reply(body=_response(raw_id, "queued"))
assert request.body in (b"", b"{}")
return Reply(body=_response(raw_id, "cancelled"))
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
client: Final = _client(gateway, gateway.key, prefix)
created: Final = client.responses.create(model=model, input="cancel lifecycle", background=True)
cancelled: Final = client.responses.cancel(created.id)
assert cancelled.id != raw_id and same_response(cancelled.id, created.id), cancelled.model_dump_json()
assert cancelled.model_dump(mode="json", exclude_none=True) == {
"id": cancelled.id,
"object": "response",
"created_at": 1.0,
"status": "cancelled",
"model": _MODEL,
"output": [],
}, cancelled.model_dump_json()
requests: Final = drain_contract_requests(wire)
assert [(request.method, request.target) for request in requests] == [
("POST", "/v1/responses"),
("POST", f"/v1/responses/{raw_id}/cancel"),
], requests
assert requests[1].body == b"{}", requests[1]
@pytest.mark.parametrize("prefix", _ALIASES)
def test_input_items_pagination_query_reaches_the_provider(gateway: Gateway, prefix: str) -> None:
pytest.skip(
"BUG: GET /v1/responses/{id}/input_items drops pagination and include query params; "
"upstream receives limit=20&order=desc without after or include"
)
raw_id: Final = f"resp_input_items_{uuid.uuid4().hex}"
def respond(request: Request) -> Reply:
if request.method == "GET" and request.target == "/v1/models":
return model_discovery_reply(_MODEL)
if request.method == "POST":
return Reply(body=_response(raw_id))
return Reply(body=b'{"data":[],"has_more":false,"object":"list"}')
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire)
client: Final = _client(gateway, gateway.key, prefix)
created: Final = client.responses.create(model=model, input="list response input items")
page: Final = client.responses.input_items.list(
created.id,
limit=2,
order="asc",
after="item_after",
include=["message.output_text.logprobs"],
)
assert page.data == [], page.model_dump_json()
requests: Final = drain_contract_requests(wire)
assert [(request.method, request.target) for request in requests] == [
("POST", "/v1/responses"),
(
"GET",
f"/v1/responses/{raw_id}/input_items?limit=2&order=asc&after=item_after"
"&include=message.output_text.logprobs",
),
], requests
assert all(request.headers["authorization"] == f"Bearer {_API_KEY}" for request in requests), requests
assert requests[1].body == b"", requests[1]