mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-07 02:59:05 +00:00
test(anthropic): drop responses bridge Claude Code tests to keep this PR Anthropic direct only
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
kerry
parent
90c8f22174
commit
b93c26a789
9 changed files with 0 additions and 1095 deletions
|
|
@ -8,8 +8,6 @@ from pydantic import JsonValue, TypeAdapter
|
|||
|
||||
JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
|
||||
ANTHROPIC_API_KEY: Final = "synthetic-anthropic-key"
|
||||
OPENAI_API_KEY: Final = "synthetic-openai-key"
|
||||
OPENAI_BACKEND: Final = "gpt-5.4-mini"
|
||||
SONNET: Final = "claude-sonnet-4-5"
|
||||
FABLE: Final = "claude-fable-5-1"
|
||||
OPUS: Final = "claude-opus-5-5"
|
||||
|
|
@ -772,114 +770,3 @@ def tool_use_stream(
|
|||
sse_frame("message_stop", {"type": "message_stop"}),
|
||||
]
|
||||
return tuple(frames)
|
||||
|
||||
|
||||
def responses_completed(
|
||||
identity: str,
|
||||
model: str,
|
||||
output_items: tuple[dict[str, JsonValue], ...],
|
||||
usage: dict[str, int],
|
||||
status: str = "completed",
|
||||
incomplete_details: JsonValue = None,
|
||||
) -> bytes:
|
||||
return json.dumps(
|
||||
{
|
||||
"id": f"resp_{identity}",
|
||||
"object": "response",
|
||||
"created_at": 1789788253,
|
||||
"status": status,
|
||||
"incomplete_details": incomplete_details,
|
||||
"model": model,
|
||||
"output": list(output_items),
|
||||
"usage": usage,
|
||||
}
|
||||
).encode()
|
||||
|
||||
|
||||
def responses_stream(identity: str, model: str, output_items: tuple[dict[str, JsonValue], ...]) -> tuple[bytes, ...]:
|
||||
frames: list[bytes] = [
|
||||
sse_frame(
|
||||
"response.created",
|
||||
{
|
||||
"type": "response.created",
|
||||
"response": {
|
||||
"id": f"resp_{identity}",
|
||||
"object": "response",
|
||||
"status": "in_progress",
|
||||
"model": model,
|
||||
"output": [],
|
||||
},
|
||||
},
|
||||
)
|
||||
]
|
||||
for index, item in enumerate(output_items):
|
||||
item_id: Final = str(item.get("id", f"item_{index}"))
|
||||
frames.append(
|
||||
sse_frame(
|
||||
"response.output_item.added",
|
||||
{
|
||||
"type": "response.output_item.added",
|
||||
"output_index": index,
|
||||
"item": {**item, "content": []} if item.get("type") == "message" else item,
|
||||
},
|
||||
)
|
||||
)
|
||||
if item.get("type") == "message":
|
||||
text: Final = "".join(part.get("text", "") for part in item.get("content", ()) if isinstance(part, dict))
|
||||
frames.append(
|
||||
sse_frame(
|
||||
"response.output_text.delta",
|
||||
{"type": "response.output_text.delta", "output_index": index, "item_id": item_id, "delta": text},
|
||||
)
|
||||
)
|
||||
if item.get("type") == "reasoning":
|
||||
summary_text: Final = "".join(
|
||||
str(part.get("text", "")) for part in item.get("summary", ()) if isinstance(part, dict)
|
||||
)
|
||||
if summary_text:
|
||||
frames.append(
|
||||
sse_frame(
|
||||
"response.reasoning_summary_text.delta",
|
||||
{
|
||||
"type": "response.reasoning_summary_text.delta",
|
||||
"output_index": index,
|
||||
"item_id": item_id,
|
||||
"delta": summary_text,
|
||||
},
|
||||
)
|
||||
)
|
||||
if item.get("type") == "function_call":
|
||||
frames.append(
|
||||
sse_frame(
|
||||
"response.function_call_arguments.delta",
|
||||
{
|
||||
"type": "response.function_call_arguments.delta",
|
||||
"output_index": index,
|
||||
"item_id": item_id,
|
||||
"delta": item.get("arguments", ""),
|
||||
},
|
||||
)
|
||||
)
|
||||
frames.append(
|
||||
sse_frame(
|
||||
"response.output_item.done",
|
||||
{"type": "response.output_item.done", "output_index": index, "item": item},
|
||||
)
|
||||
)
|
||||
frames.append(
|
||||
sse_frame(
|
||||
"response.completed",
|
||||
{
|
||||
"type": "response.completed",
|
||||
"response": {
|
||||
"id": f"resp_{identity}",
|
||||
"object": "response",
|
||||
"status": "completed",
|
||||
"model": model,
|
||||
"output": list(output_items),
|
||||
"usage": {"input_tokens": 41, "output_tokens": 5, "total_tokens": 46},
|
||||
},
|
||||
},
|
||||
)
|
||||
)
|
||||
return tuple(frames)
|
||||
|
|
|
|||
|
|
@ -1,107 +0,0 @@
|
|||
import uuid
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from integration._support.client import Gateway
|
||||
from integration._support.wire import Reply, Request, wire_server
|
||||
from integration.messages_endpoint import _claude_code as cc
|
||||
|
||||
|
||||
def test_compact_edit_maps_to_responses_context_management(gateway: Gateway) -> None:
|
||||
request_body: Final = {
|
||||
**cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000),
|
||||
"context_management": {
|
||||
"edits": [
|
||||
{"type": "clear_thinking_20251015", "keep": "all"},
|
||||
{"type": "compact_20260112", "trigger": {"type": "input_tokens", "value": 150000}},
|
||||
]
|
||||
},
|
||||
"stream": False,
|
||||
}
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
assert request.method == "POST"
|
||||
assert request.target == "/responses", request.target
|
||||
body: Final = cc.JSON_OBJECT.validate_json(request.body)
|
||||
assert body.get("context_management") == [{"type": "compaction", "compact_threshold": 150000}], body.get(
|
||||
"context_management"
|
||||
)
|
||||
return Reply(
|
||||
body=cc.responses_completed(
|
||||
"cm",
|
||||
cc.OPENAI_BACKEND,
|
||||
(
|
||||
{
|
||||
"type": "message",
|
||||
"id": "msg_1",
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "OK", "annotations": []}],
|
||||
},
|
||||
),
|
||||
{"input_tokens": 41, "output_tokens": 3, "total_tokens": 44},
|
||||
)
|
||||
)
|
||||
|
||||
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model=f"openai/{cc.OPENAI_BACKEND}", api_base=wire.url, api_key=cc.OPENAI_API_KEY)
|
||||
response: Final = gateway.request(
|
||||
"POST",
|
||||
"/v1/messages",
|
||||
{**request_body, "model": model},
|
||||
params={"beta": "true"},
|
||||
headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA),
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
payload: Final = cc.JSON_OBJECT.validate_json(response.content)
|
||||
assert payload["content"] == [{"type": "text", "text": "OK"}], payload["content"]
|
||||
assert len(wire.drain()) == 1
|
||||
|
||||
|
||||
def test_compaction_output_item_reaches_client_as_compaction_block(gateway: Gateway) -> None:
|
||||
pytest.skip(
|
||||
"BUG: the responses bridge drops compaction output items in translate_response "
|
||||
"(transformation.py handles only message/reasoning/function_call), so the client loses "
|
||||
"the compaction block entirely"
|
||||
)
|
||||
request_body: Final = {
|
||||
**cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000),
|
||||
"context_management": {
|
||||
"edits": [{"type": "compact_20260112", "trigger": {"type": "input_tokens", "value": 150000}}]
|
||||
},
|
||||
"stream": False,
|
||||
}
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
return Reply(
|
||||
body=cc.responses_completed(
|
||||
"cm",
|
||||
cc.OPENAI_BACKEND,
|
||||
(
|
||||
{"type": "compaction", "content": "<summary>"},
|
||||
{
|
||||
"type": "message",
|
||||
"id": "msg_1",
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "OK", "annotations": []}],
|
||||
},
|
||||
),
|
||||
{"input_tokens": 41, "output_tokens": 3, "total_tokens": 44},
|
||||
)
|
||||
)
|
||||
|
||||
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model=f"openai/{cc.OPENAI_BACKEND}", api_base=wire.url, api_key=cc.OPENAI_API_KEY)
|
||||
response: Final = gateway.request(
|
||||
"POST",
|
||||
"/v1/messages",
|
||||
{**request_body, "model": model},
|
||||
params={"beta": "true"},
|
||||
headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA),
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
payload: Final = cc.JSON_OBJECT.validate_json(response.content)
|
||||
compaction_blocks: Final = [block for block in payload["content"] if block.get("type") == "compaction"]
|
||||
assert compaction_blocks == [{"type": "compaction", "content": "<summary>"}], payload["content"]
|
||||
assert len(wire.drain()) == 1
|
||||
|
|
@ -1,73 +0,0 @@
|
|||
import base64
|
||||
import uuid
|
||||
from typing import Final
|
||||
|
||||
from integration._support.client import Gateway
|
||||
from integration._support.wire import Reply, Request, wire_server
|
||||
from integration.messages_endpoint import _claude_code as cc
|
||||
|
||||
_PDF_BYTES: Final = (
|
||||
b"%PDF-1.1\n"
|
||||
b"1 0 obj<</Type/Catalog/Pages 2 0 R>>endobj\n"
|
||||
b"2 0 obj<</Type/Pages/Kids[3 0 R]/Count 1>>endobj\n"
|
||||
b"3 0 obj<</Type/Page/Parent 2 0 R/MediaBox[0 0 72 72]>>endobj\n"
|
||||
b"trailer<</Root 1 0 R>>\n%%EOF"
|
||||
)
|
||||
_PDF_B64: Final = base64.b64encode(_PDF_BYTES).decode()
|
||||
|
||||
|
||||
def test_pdf_document_block_maps_to_input_file_on_bridge(gateway: Gateway) -> None:
|
||||
request_body: Final = {**cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000), "stream": False}
|
||||
doc_text: Final = f"What is on page one? {uuid.uuid4().hex}"
|
||||
request_body["messages"] = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{
|
||||
"type": "document",
|
||||
"source": {"type": "base64", "data": _PDF_B64, "media_type": "application/pdf"},
|
||||
"title": "dot.pdf",
|
||||
},
|
||||
{"type": "text", "text": doc_text},
|
||||
],
|
||||
}
|
||||
]
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
assert request.target == "/responses", request.target
|
||||
body: Final = cc.JSON_OBJECT.validate_json(request.body)
|
||||
user_msg: Final = body["input"][0]
|
||||
assert user_msg["content"][0] == {
|
||||
"type": "input_file",
|
||||
"filename": "dot.pdf",
|
||||
"file_data": f"data:application/pdf;base64,{_PDF_B64}",
|
||||
}, user_msg
|
||||
assert user_msg["content"][1] == {"type": "input_text", "text": doc_text}, user_msg
|
||||
return Reply(
|
||||
body=cc.responses_completed(
|
||||
"doc",
|
||||
cc.OPENAI_BACKEND,
|
||||
(
|
||||
{
|
||||
"type": "message",
|
||||
"id": "msg_1",
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "Page one.", "annotations": []}],
|
||||
},
|
||||
),
|
||||
{"input_tokens": 41, "output_tokens": 3, "total_tokens": 44},
|
||||
)
|
||||
)
|
||||
|
||||
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model=f"openai/{cc.OPENAI_BACKEND}", api_base=wire.url, api_key=cc.OPENAI_API_KEY)
|
||||
response: Final = gateway.request(
|
||||
"POST",
|
||||
"/v1/messages",
|
||||
{**request_body, "model": model},
|
||||
params={"beta": "true"},
|
||||
headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA),
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
assert len(wire.drain()) == 1
|
||||
|
|
@ -1,149 +0,0 @@
|
|||
import uuid
|
||||
from typing import Final
|
||||
|
||||
from integration._support.client import Gateway
|
||||
from integration._support.wire import Reply, Request, wire_server
|
||||
from integration.messages_endpoint import _claude_code as cc
|
||||
from pydantic import JsonValue
|
||||
|
||||
_INSTRUCTIONS: Final = "\n".join(block["text"] for block in cc.system_blocks())
|
||||
_OUTPUT_ITEMS: Final = (
|
||||
{
|
||||
"type": "reasoning",
|
||||
"id": "rs_1",
|
||||
"summary": [{"type": "summary_text", "text": "short plan"}],
|
||||
"encrypted_content": "enc_1",
|
||||
},
|
||||
{
|
||||
"type": "message",
|
||||
"id": "msg_1",
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "PONG", "annotations": []}],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _expected_responses_body(request_body: dict[str, JsonValue], effort: str) -> dict[str, JsonValue]:
|
||||
user_blocks: Final = request_body["messages"][0]["content"]
|
||||
expected_tools: Final = tuple(
|
||||
{
|
||||
"type": "function",
|
||||
"name": tool["name"],
|
||||
"strict": False,
|
||||
"description": tool["description"],
|
||||
"parameters": tool["input_schema"],
|
||||
}
|
||||
for tool in request_body["tools"]
|
||||
)
|
||||
input_items: Final = [
|
||||
{
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": block["text"]} for block in user_blocks],
|
||||
}
|
||||
]
|
||||
for message in request_body["messages"][1:]:
|
||||
input_items.append(
|
||||
{
|
||||
"type": "message",
|
||||
"role": "system",
|
||||
"content": [
|
||||
{"type": "input_text", "text": block["text"]}
|
||||
for block in message["content"]
|
||||
if block.get("type") == "text"
|
||||
],
|
||||
}
|
||||
)
|
||||
return {
|
||||
"model": cc.OPENAI_BACKEND,
|
||||
"input": input_items,
|
||||
"include": ["reasoning.encrypted_content"],
|
||||
"instructions": _INSTRUCTIONS,
|
||||
"max_output_tokens": request_body["max_tokens"],
|
||||
"tools": list(expected_tools),
|
||||
"reasoning": {"effort": effort},
|
||||
"stream": True,
|
||||
"user": cc.METADATA_USER_ID[:64],
|
||||
"prompt_cache_key": "00000000-0000-4000-8000-000000000000",
|
||||
}
|
||||
|
||||
|
||||
def _assert_client_events(text: str) -> None:
|
||||
events: Final = cc.sse_events(text)
|
||||
assert [event for event, _ in events] == [
|
||||
"message_start",
|
||||
"content_block_start",
|
||||
"content_block_delta",
|
||||
"content_block_delta",
|
||||
"content_block_stop",
|
||||
"content_block_start",
|
||||
"content_block_delta",
|
||||
"content_block_stop",
|
||||
"message_delta",
|
||||
"message_stop",
|
||||
], [event for event, _ in events]
|
||||
assert events[1][1]["content_block"]["type"] == "thinking"
|
||||
assert events[2][1]["delta"] == {"type": "thinking_delta", "thinking": "short plan"}
|
||||
assert events[3][1]["delta"]["type"] == "signature_delta"
|
||||
assert events[6][1]["delta"] == {"type": "text_delta", "text": "PONG"}
|
||||
assert events[8][1]["delta"]["stop_reason"] == "end_turn"
|
||||
|
||||
|
||||
def test_claude_code_frontier_body_becomes_reasoning_effort_on_responses_bridge(gateway: Gateway) -> None:
|
||||
request_body: Final = cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000)
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
assert request.method == "POST"
|
||||
assert request.target == "/responses", request.target
|
||||
assert request.headers["authorization"] == f"Bearer {cc.OPENAI_API_KEY}"
|
||||
body: Final = cc.JSON_OBJECT.validate_json(request.body)
|
||||
expected: Final = _expected_responses_body(request_body, "high")
|
||||
assert body == expected, {
|
||||
key: {"expected": expected.get(key), "upstream": body.get(key)}
|
||||
for key in expected.keys() | body.keys()
|
||||
if expected.get(key) != body.get(key)
|
||||
}
|
||||
return Reply(
|
||||
content_type="text/event-stream", chunks=cc.responses_stream("bridge1", cc.OPENAI_BACKEND, _OUTPUT_ITEMS)
|
||||
)
|
||||
|
||||
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model=f"openai/{cc.OPENAI_BACKEND}", api_base=wire.url, api_key=cc.OPENAI_API_KEY)
|
||||
response: Final = gateway.request(
|
||||
"POST",
|
||||
"/v1/messages",
|
||||
{**request_body, "model": model},
|
||||
params={"beta": "true"},
|
||||
headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA),
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
_assert_client_events(response.text)
|
||||
assert len(wire.drain()) == 1
|
||||
|
||||
|
||||
def test_claude_code_legacy_thinking_budget_maps_to_reasoning_effort_on_bridge(gateway: Gateway) -> None:
|
||||
request_body: Final = cc.claude_code_request(f"cache-bust-{uuid.uuid4().hex}")
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
assert request.method == "POST"
|
||||
assert request.target == "/responses", request.target
|
||||
body: Final = cc.JSON_OBJECT.validate_json(request.body)
|
||||
assert body["reasoning"] == {"effort": "high"}, body.get("reasoning")
|
||||
assert body["model"] == cc.OPENAI_BACKEND
|
||||
return Reply(
|
||||
content_type="text/event-stream", chunks=cc.responses_stream("bridge2", cc.OPENAI_BACKEND, _OUTPUT_ITEMS)
|
||||
)
|
||||
|
||||
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model=f"openai/{cc.OPENAI_BACKEND}", api_base=wire.url, api_key=cc.OPENAI_API_KEY)
|
||||
response: Final = gateway.request(
|
||||
"POST",
|
||||
"/v1/messages",
|
||||
{**request_body, "model": model},
|
||||
params={"beta": "true"},
|
||||
headers=cc.cli_headers(gateway.key),
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
_assert_client_events(response.text)
|
||||
assert len(wire.drain()) == 1
|
||||
|
|
@ -1,112 +0,0 @@
|
|||
import uuid
|
||||
from typing import Final
|
||||
|
||||
from integration._support.client import Gateway
|
||||
from integration._support.wire import Reply, Request, wire_server
|
||||
from integration.messages_endpoint import _claude_code as cc
|
||||
from pydantic import JsonValue
|
||||
|
||||
_PNG_B64: Final = "iVBORw0KGgoAAAANSUhEUgAAAAQAAAAECAIAAAAmkwkpAAAAEElEQVR4nGP4z8AARwzEcQCukw/x0F8jngAAAABJRU5ErkJggg=="
|
||||
_IMAGE_BLOCK: Final = {
|
||||
"type": "image",
|
||||
"source": {"type": "base64", "data": _PNG_B64, "media_type": "image/png"},
|
||||
}
|
||||
_DATA_URL: Final = f"data:image/png;base64,{_PNG_B64}"
|
||||
|
||||
|
||||
def _respond_ok(request: Request) -> Reply:
|
||||
return Reply(
|
||||
body=cc.responses_completed(
|
||||
"img",
|
||||
cc.OPENAI_BACKEND,
|
||||
(
|
||||
{
|
||||
"type": "message",
|
||||
"id": "msg_1",
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "RED", "annotations": []}],
|
||||
},
|
||||
),
|
||||
{"input_tokens": 41, "output_tokens": 3, "total_tokens": 44},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def test_tool_result_image_maps_to_input_image_on_bridge(gateway: Gateway) -> None:
|
||||
turn1: Final = {
|
||||
**cc.frontier_request(
|
||||
f"cache-bust-{uuid.uuid4().hex}",
|
||||
"high",
|
||||
64000,
|
||||
prompt_text="Read /tmp/cc_probe/dot.png and say what colour it is",
|
||||
),
|
||||
"stream": False,
|
||||
}
|
||||
turn2: Final = cc.tool_loop_turn2(
|
||||
turn1,
|
||||
({"type": "tool_use", "id": "call_img", "name": "Read", "input": {"file_path": "/tmp/cc_probe/dot.png"}},),
|
||||
(("call_img", [dict(_IMAGE_BLOCK)]),),
|
||||
)
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
body: Final = cc.JSON_OBJECT.validate_json(request.body)
|
||||
outputs: Final = [
|
||||
item for item in body["input"] if isinstance(item, dict) and item.get("type") == "function_call_output"
|
||||
]
|
||||
assert len(outputs) == 1 and outputs[0]["call_id"] == "call_img", outputs
|
||||
image_messages: Final = [
|
||||
item
|
||||
for item in body["input"]
|
||||
if isinstance(item, dict)
|
||||
and item.get("type") == "message"
|
||||
and item.get("role") == "user"
|
||||
and any(isinstance(part, dict) and part.get("type") == "input_image" for part in item.get("content", ()))
|
||||
]
|
||||
assert image_messages, body["input"]
|
||||
image_parts: Final = [
|
||||
part
|
||||
for part in image_messages[0]["content"]
|
||||
if isinstance(part, dict) and part.get("type") == "input_image"
|
||||
]
|
||||
assert image_parts[0]["image_url"] == _DATA_URL, image_parts
|
||||
return _respond_ok(request)
|
||||
|
||||
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model=f"openai/{cc.OPENAI_BACKEND}", api_base=wire.url, api_key=cc.OPENAI_API_KEY)
|
||||
response: Final = gateway.request(
|
||||
"POST",
|
||||
"/v1/messages",
|
||||
{**turn2, "model": model},
|
||||
params={"beta": "true"},
|
||||
headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA),
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
assert len(wire.drain()) == 1
|
||||
|
||||
|
||||
def test_pasted_image_maps_to_input_image_on_bridge(gateway: Gateway) -> None:
|
||||
request_body: Final = {**cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000), "stream": False}
|
||||
pasted_text: Final = f"What colour is this? {uuid.uuid4().hex}"
|
||||
request_body["messages"] = [
|
||||
{"role": "user", "content": [dict(_IMAGE_BLOCK), {"type": "text", "text": pasted_text}]}
|
||||
]
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
body: Final = cc.JSON_OBJECT.validate_json(request.body)
|
||||
user_msg: Final = body["input"][0]
|
||||
assert user_msg["content"][0] == {"type": "input_image", "image_url": _DATA_URL}, user_msg
|
||||
assert user_msg["content"][1] == {"type": "input_text", "text": pasted_text}, user_msg
|
||||
return _respond_ok(request)
|
||||
|
||||
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model=f"openai/{cc.OPENAI_BACKEND}", api_base=wire.url, api_key=cc.OPENAI_API_KEY)
|
||||
response: Final = gateway.request(
|
||||
"POST",
|
||||
"/v1/messages",
|
||||
{**request_body, "model": model},
|
||||
params={"beta": "true"},
|
||||
headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA),
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
assert len(wire.drain()) == 1
|
||||
|
|
@ -1,128 +0,0 @@
|
|||
import uuid
|
||||
from typing import Final
|
||||
|
||||
from integration._support.client import Gateway
|
||||
from integration._support.wire import Reply, Request, wire_server
|
||||
from integration.messages_endpoint import _claude_code as cc
|
||||
from pydantic import JsonValue
|
||||
|
||||
|
||||
def _expected_input(turn1: dict[str, JsonValue]) -> list[JsonValue]:
|
||||
return [
|
||||
{
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": block["text"]} for block in turn1["messages"][0]["content"]],
|
||||
},
|
||||
{
|
||||
"type": "message",
|
||||
"role": "system",
|
||||
"content": [{"type": "input_text", "text": block["text"]} for block in turn1["messages"][1]["content"]],
|
||||
},
|
||||
{"type": "reasoning", "summary": [{"type": "summary_text", "text": "plan"}]},
|
||||
{
|
||||
"type": "function_call",
|
||||
"call_id": "toolu_a",
|
||||
"name": "Read",
|
||||
"arguments": '{"file_path": "/tmp/cc_probe/a.txt"}',
|
||||
},
|
||||
{"type": "function_call_output", "call_id": "toolu_a", "output": "ALPHA"},
|
||||
{
|
||||
"type": "message",
|
||||
"role": "system",
|
||||
"content": [
|
||||
{"type": "input_text", "text": "<total_tokens>14999970 tokens left</total_tokens>"},
|
||||
{
|
||||
"type": "input_text",
|
||||
"text": "First privately list what you need next; then request every item that doesn't depend on another's result in this one response.",
|
||||
},
|
||||
],
|
||||
},
|
||||
{"type": "reasoning", "summary": [{"type": "summary_text", "text": "got A"}]},
|
||||
{
|
||||
"type": "function_call",
|
||||
"call_id": "toolu_b",
|
||||
"name": "Read",
|
||||
"arguments": '{"file_path": "/tmp/cc_probe/b.txt"}',
|
||||
},
|
||||
{
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "got A"}],
|
||||
},
|
||||
{"type": "function_call_output", "call_id": "toolu_b", "output": "BRAVO"},
|
||||
{
|
||||
"type": "message",
|
||||
"role": "system",
|
||||
"content": [
|
||||
{"type": "input_text", "text": "<total_tokens>14999970 tokens left</total_tokens>"},
|
||||
{
|
||||
"type": "input_text",
|
||||
"text": "First privately list what you need next; then request every item that doesn't depend on another's result in this one response.",
|
||||
},
|
||||
],
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def test_bridge_replays_interleaved_history_in_order(gateway: Gateway) -> None:
|
||||
turn1: Final = {
|
||||
**cc.frontier_request(
|
||||
f"cache-bust-{uuid.uuid4().hex}",
|
||||
"high",
|
||||
64000,
|
||||
prompt_text="Read /tmp/cc_probe/a.txt then /tmp/cc_probe/b.txt one at a time",
|
||||
),
|
||||
"stream": False,
|
||||
}
|
||||
turn2: Final = cc.tool_loop_turn2(
|
||||
turn1,
|
||||
(
|
||||
{"type": "thinking", "thinking": "plan", "signature": "sig_anthropic_1"},
|
||||
{"type": "tool_use", "id": "toolu_a", "name": "Read", "input": {"file_path": "/tmp/cc_probe/a.txt"}},
|
||||
),
|
||||
(("toolu_a", "ALPHA"),),
|
||||
)
|
||||
turn3: Final = cc.tool_loop_turn2(
|
||||
turn2,
|
||||
(
|
||||
{"type": "thinking", "thinking": "got A", "signature": "sig_anthropic_2"},
|
||||
{"type": "text", "text": "got A"},
|
||||
{"type": "tool_use", "id": "toolu_b", "name": "Read", "input": {"file_path": "/tmp/cc_probe/b.txt"}},
|
||||
),
|
||||
(("toolu_b", "BRAVO"),),
|
||||
)
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
assert request.method == "POST"
|
||||
assert request.target == "/responses", request.target
|
||||
body: Final = cc.JSON_OBJECT.validate_json(request.body)
|
||||
assert body["input"] == _expected_input(turn1), body["input"]
|
||||
return Reply(
|
||||
body=cc.responses_completed(
|
||||
"il",
|
||||
cc.OPENAI_BACKEND,
|
||||
(
|
||||
{
|
||||
"type": "message",
|
||||
"id": "msg_1",
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "ALPHA BRAVO", "annotations": []}],
|
||||
},
|
||||
),
|
||||
{"input_tokens": 50, "output_tokens": 4, "total_tokens": 54},
|
||||
)
|
||||
)
|
||||
|
||||
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model=f"openai/{cc.OPENAI_BACKEND}", api_base=wire.url, api_key=cc.OPENAI_API_KEY)
|
||||
response: Final = gateway.request(
|
||||
"POST",
|
||||
"/v1/messages",
|
||||
{**turn3, "model": model},
|
||||
params={"beta": "true"},
|
||||
headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA),
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
assert len(wire.drain()) == 1
|
||||
|
|
@ -1,95 +0,0 @@
|
|||
import uuid
|
||||
from typing import Final
|
||||
|
||||
from integration._support.client import Gateway
|
||||
from integration._support.wire import Reply, Request, wire_server
|
||||
from integration.messages_endpoint import _claude_code as cc
|
||||
from pydantic import JsonValue
|
||||
|
||||
|
||||
def test_anthropic_signed_thinking_in_history_crosses_to_responses_bridge(gateway: Gateway) -> None:
|
||||
turn1: Final = cc.frontier_request(
|
||||
f"cache-bust-{uuid.uuid4().hex}",
|
||||
"high",
|
||||
64000,
|
||||
prompt_text="Read /tmp/cc_probe/hello.txt and reply with its single word",
|
||||
)
|
||||
turn2: Final = cc.tool_loop_turn2(
|
||||
turn1,
|
||||
(
|
||||
{"type": "thinking", "thinking": "need to read the file", "signature": "sig_anthropic_1"},
|
||||
{
|
||||
"type": "tool_use",
|
||||
"id": "toolu_read_1",
|
||||
"name": "Read",
|
||||
"input": {"file_path": "/tmp/cc_probe/hello.txt"},
|
||||
},
|
||||
),
|
||||
(("toolu_read_1", "1\tPROBE\n2\t"),),
|
||||
)
|
||||
bridge_input_box: list[JsonValue] = []
|
||||
|
||||
def respond_anthropic(request: Request) -> Reply:
|
||||
assert request.target == "/v1/messages", request.target
|
||||
return Reply(
|
||||
content_type="text/event-stream",
|
||||
chunks=cc.tool_use_stream(
|
||||
f"msg_sw_{uuid.uuid4().hex}",
|
||||
cc.FABLE,
|
||||
"need to read the file",
|
||||
"sig_anthropic_1",
|
||||
(("toolu_read_1", "Read", {"file_path": "/tmp/cc_probe/hello.txt"}),),
|
||||
{"input_tokens": 20, "output_tokens": 10},
|
||||
),
|
||||
)
|
||||
|
||||
def respond_openai(request: Request) -> Reply:
|
||||
assert request.method == "POST"
|
||||
assert request.target == "/responses", request.target
|
||||
body: Final = cc.JSON_OBJECT.validate_json(request.body)
|
||||
bridge_input_box.append(body.get("input"))
|
||||
reasoning_items: Final = [
|
||||
item for item in body["input"] if isinstance(item, dict) and item.get("type") == "reasoning"
|
||||
]
|
||||
assert reasoning_items == [
|
||||
{"type": "reasoning", "summary": [{"type": "summary_text", "text": "need to read the file"}]}
|
||||
], reasoning_items
|
||||
return Reply(
|
||||
content_type="text/event-stream",
|
||||
chunks=cc.responses_stream(
|
||||
"sw",
|
||||
cc.OPENAI_BACKEND,
|
||||
(
|
||||
{
|
||||
"type": "message",
|
||||
"id": "msg_1",
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "PROBE", "annotations": []}],
|
||||
},
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
with (
|
||||
wire_server(respond_anthropic) as wire_a,
|
||||
wire_server(respond_openai) as wire_b,
|
||||
gateway.scenario() as scenario,
|
||||
):
|
||||
fable: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire_a.url, api_key=cc.ANTHROPIC_API_KEY)
|
||||
openai_alias: Final = scenario.model(
|
||||
model=f"openai/{cc.OPENAI_BACKEND}", api_base=wire_b.url, api_key=cc.OPENAI_API_KEY
|
||||
)
|
||||
headers: Final = cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA)
|
||||
response1: Final = gateway.request(
|
||||
"POST", "/v1/messages", {**turn1, "model": fable}, params={"beta": "true"}, headers=headers
|
||||
)
|
||||
assert response1.status_code == 200, response1.text
|
||||
response2: Final = gateway.request(
|
||||
"POST", "/v1/messages", {**turn2, "model": openai_alias}, params={"beta": "true"}, headers=headers
|
||||
)
|
||||
assert response2.status_code == 200, response2.text
|
||||
events: Final = cc.sse_events(response2.text)
|
||||
assert events[-1][0] == "message_stop"
|
||||
assert len(wire_a.drain()) == 1
|
||||
assert len(wire_b.drain()) == 1
|
||||
|
|
@ -1,223 +0,0 @@
|
|||
import json
|
||||
import uuid
|
||||
from typing import Final
|
||||
|
||||
from integration._support.client import Gateway
|
||||
from integration._support.wire import Reply, Request, wire_server
|
||||
from integration.messages_endpoint import _claude_code as cc
|
||||
from pydantic import JsonValue
|
||||
|
||||
_INSTRUCTIONS: Final = "\n".join(block["text"] for block in cc.system_blocks())
|
||||
|
||||
|
||||
def _turn1() -> dict[str, JsonValue]:
|
||||
return {
|
||||
**cc.frontier_request(
|
||||
f"cache-bust-{uuid.uuid4().hex}",
|
||||
"high",
|
||||
64000,
|
||||
prompt_text="Read /tmp/cc_probe/hello.txt and reply with its single word",
|
||||
),
|
||||
"stream": False,
|
||||
}
|
||||
|
||||
|
||||
def _upstream_items(calls: tuple[tuple[str, str, JsonValue], ...]) -> tuple[dict[str, JsonValue], ...]:
|
||||
return (
|
||||
{
|
||||
"type": "reasoning",
|
||||
"id": "rs_1",
|
||||
"summary": [{"type": "summary_text", "text": "short plan"}],
|
||||
"encrypted_content": "enc_1",
|
||||
},
|
||||
*(
|
||||
{
|
||||
"type": "function_call",
|
||||
"id": f"fc_{i}",
|
||||
"call_id": call_id,
|
||||
"name": name,
|
||||
"arguments": json.dumps(tool_input),
|
||||
"status": "completed",
|
||||
}
|
||||
for i, (call_id, name, tool_input) in enumerate(calls, start=1)
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _expected_turn2_input(
|
||||
turn1: dict[str, JsonValue],
|
||||
assistant_content: tuple[dict[str, JsonValue], ...],
|
||||
tool_results: tuple[tuple[str, JsonValue], ...],
|
||||
) -> list[JsonValue]:
|
||||
items: Final = [
|
||||
{
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": block["text"]} for block in turn1["messages"][0]["content"]],
|
||||
},
|
||||
{
|
||||
"type": "message",
|
||||
"role": "system",
|
||||
"content": [{"type": "input_text", "text": block["text"]} for block in turn1["messages"][1]["content"]],
|
||||
},
|
||||
{
|
||||
"type": "reasoning",
|
||||
"summary": [{"type": "summary_text", "text": "short plan"}],
|
||||
"encrypted_content": "enc_1",
|
||||
},
|
||||
]
|
||||
items += [
|
||||
{
|
||||
"type": "function_call",
|
||||
"call_id": block["id"],
|
||||
"name": block["name"],
|
||||
"arguments": json.dumps(block["input"]),
|
||||
}
|
||||
for block in assistant_content
|
||||
if block.get("type") == "tool_use"
|
||||
]
|
||||
items += [
|
||||
{"type": "function_call_output", "call_id": tool_use_id, "output": content}
|
||||
for tool_use_id, content in tool_results
|
||||
]
|
||||
items.append(
|
||||
{
|
||||
"type": "message",
|
||||
"role": "system",
|
||||
"content": [
|
||||
{"type": "input_text", "text": "<total_tokens>14999970 tokens left</total_tokens>"},
|
||||
{
|
||||
"type": "input_text",
|
||||
"text": "First privately list what you need next; then request every item that doesn't depend on another's result in this one response.",
|
||||
},
|
||||
],
|
||||
}
|
||||
)
|
||||
return items
|
||||
|
||||
|
||||
def test_responses_bridge_replays_reasoning_and_tool_call_on_turn_two(gateway: Gateway) -> None:
|
||||
turn1: Final = _turn1()
|
||||
calls: Final = (("call_1", "Read", {"file_path": "/tmp/cc_probe/hello.txt"}),)
|
||||
tool_results: Final = (("call_1", "1\tPROBE\n2\t"),)
|
||||
seen: list[dict[str, JsonValue]] = []
|
||||
signature_box: list[str] = []
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
assert request.method == "POST"
|
||||
assert request.target == "/responses", request.target
|
||||
body: Final = cc.JSON_OBJECT.validate_json(request.body)
|
||||
seen.append(body)
|
||||
if len(seen) == 1:
|
||||
return Reply(
|
||||
body=cc.responses_completed(
|
||||
"tl1",
|
||||
cc.OPENAI_BACKEND,
|
||||
_upstream_items(calls),
|
||||
{"input_tokens": 41, "output_tokens": 5, "total_tokens": 46},
|
||||
)
|
||||
)
|
||||
assistant_content: Final = (
|
||||
{"type": "thinking", "thinking": "short plan", "signature": signature_box[0]},
|
||||
{"type": "tool_use", "id": "call_1", "name": "Read", "input": {"file_path": "/tmp/cc_probe/hello.txt"}},
|
||||
)
|
||||
assert body["input"] == _expected_turn2_input(turn1, assistant_content, tool_results), body["input"]
|
||||
return Reply(
|
||||
body=cc.responses_completed(
|
||||
"tl2",
|
||||
cc.OPENAI_BACKEND,
|
||||
(
|
||||
{
|
||||
"type": "message",
|
||||
"id": "msg_1",
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "PROBE", "annotations": []}],
|
||||
},
|
||||
),
|
||||
{"input_tokens": 50, "output_tokens": 3, "total_tokens": 53},
|
||||
)
|
||||
)
|
||||
|
||||
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model=f"openai/{cc.OPENAI_BACKEND}", api_base=wire.url, api_key=cc.OPENAI_API_KEY)
|
||||
headers: Final = cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA)
|
||||
response1: Final = gateway.request(
|
||||
"POST", "/v1/messages", {**turn1, "model": model}, params={"beta": "true"}, headers=headers
|
||||
)
|
||||
assert response1.status_code == 200, response1.text
|
||||
payload: Final = cc.JSON_OBJECT.validate_json(response1.content)
|
||||
assert payload["stop_reason"] == "tool_use", payload
|
||||
assert payload["content"][0]["type"] == "thinking", payload["content"]
|
||||
signature: Final = payload["content"][0].get("signature")
|
||||
assert signature, payload["content"][0]
|
||||
signature_box.append(signature)
|
||||
assert payload["content"][1] == {
|
||||
"type": "tool_use",
|
||||
"id": "call_1",
|
||||
"name": "Read",
|
||||
"input": {"file_path": "/tmp/cc_probe/hello.txt"},
|
||||
}, payload["content"]
|
||||
assistant_content: Final = (
|
||||
{"type": "thinking", "thinking": "short plan", "signature": signature},
|
||||
{"type": "tool_use", "id": "call_1", "name": "Read", "input": {"file_path": "/tmp/cc_probe/hello.txt"}},
|
||||
)
|
||||
turn2: Final = cc.tool_loop_turn2(turn1, assistant_content, tool_results)
|
||||
response2: Final = gateway.request(
|
||||
"POST", "/v1/messages", {**turn2, "model": model}, params={"beta": "true"}, headers=headers
|
||||
)
|
||||
assert response2.status_code == 200, response2.text
|
||||
payload2: Final = cc.JSON_OBJECT.validate_json(response2.content)
|
||||
assert payload2["stop_reason"] == "end_turn", payload2
|
||||
assert len(wire.drain()) == 2
|
||||
|
||||
|
||||
def test_responses_bridge_replays_parallel_tool_calls_in_order(gateway: Gateway) -> None:
|
||||
turn1: Final = _turn1()
|
||||
calls: Final = (
|
||||
("call_1", "Read", {"file_path": "/tmp/cc_probe/hello.txt"}),
|
||||
("call_2", "Read", {"file_path": "/tmp/cc_probe/world.txt"}),
|
||||
)
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
body: Final = cc.JSON_OBJECT.validate_json(request.body)
|
||||
seen_items: Final = body["input"]
|
||||
outputs: Final = [
|
||||
item for item in seen_items if isinstance(item, dict) and item.get("type") == "function_call_output"
|
||||
]
|
||||
assert [item["call_id"] for item in outputs] == ["call_1", "call_2"], outputs
|
||||
return Reply(
|
||||
body=cc.responses_completed(
|
||||
"mt",
|
||||
cc.OPENAI_BACKEND,
|
||||
(
|
||||
{
|
||||
"type": "message",
|
||||
"id": "msg_1",
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "PROBE PROBE2", "annotations": []}],
|
||||
},
|
||||
),
|
||||
{"input_tokens": 50, "output_tokens": 4, "total_tokens": 54},
|
||||
)
|
||||
)
|
||||
|
||||
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model=f"openai/{cc.OPENAI_BACKEND}", api_base=wire.url, api_key=cc.OPENAI_API_KEY)
|
||||
assistant_content: Final = tuple(
|
||||
{"type": "tool_use", "id": call_id, "name": name, "input": tool_input}
|
||||
for call_id, name, tool_input in calls
|
||||
)
|
||||
turn2: Final = cc.tool_loop_turn2(
|
||||
turn1, assistant_content, (("call_1", "1\tPROBE\n2\t"), ("call_2", "1\tPROBE2\n2\t"))
|
||||
)
|
||||
response: Final = gateway.request(
|
||||
"POST",
|
||||
"/v1/messages",
|
||||
{**turn2, "model": model},
|
||||
params={"beta": "true"},
|
||||
headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA),
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
assert len(wire.drain()) == 1
|
||||
|
|
@ -1,95 +0,0 @@
|
|||
import uuid
|
||||
from typing import Final
|
||||
|
||||
from integration._support.client import Gateway
|
||||
from integration._support.wire import Reply, Request, wire_server
|
||||
from integration.messages_endpoint import _claude_code as cc
|
||||
from pydantic import JsonValue
|
||||
|
||||
_WEB_SEARCH_TOOL: Final = {
|
||||
"name": "WebSearch",
|
||||
"description": "Search the web. Returns result blocks with titles and URLs.",
|
||||
"input_schema": cc.schema(
|
||||
{
|
||||
"query": cc.field("The search query to use", type="string", minLength=2),
|
||||
"allowed_domains": cc.field(
|
||||
"Only include search results from these domains", type="array", items={"type": "string"}
|
||||
),
|
||||
"blocked_domains": cc.field(
|
||||
"Never include search results from these domains", type="array", items={"type": "string"}
|
||||
),
|
||||
},
|
||||
("query",),
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def test_web_search_tool_and_cited_output_on_responses_bridge(gateway: Gateway) -> None:
|
||||
request_body: Final = {
|
||||
**cc.frontier_request(
|
||||
f"cache-bust-{uuid.uuid4().hex}",
|
||||
"high",
|
||||
64000,
|
||||
prompt_text="Use web search to find the current LiteLLM version and answer in one word",
|
||||
),
|
||||
"stream": False,
|
||||
}
|
||||
request_body["tools"] = [*request_body["tools"], _WEB_SEARCH_TOOL]
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
assert request.method == "POST"
|
||||
assert request.target == "/responses", request.target
|
||||
body: Final = cc.JSON_OBJECT.validate_json(request.body)
|
||||
tools: Final = body["tools"]
|
||||
assert tools[-1] == {
|
||||
"type": "function",
|
||||
"name": "WebSearch",
|
||||
"strict": False,
|
||||
"description": _WEB_SEARCH_TOOL["description"],
|
||||
"parameters": _WEB_SEARCH_TOOL["input_schema"],
|
||||
}, tools[-1]
|
||||
return Reply(
|
||||
body=cc.responses_completed(
|
||||
"ws",
|
||||
cc.OPENAI_BACKEND,
|
||||
(
|
||||
{"type": "web_search_call", "id": "ws_1", "status": "completed"},
|
||||
{
|
||||
"type": "message",
|
||||
"id": "msg_1",
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [
|
||||
{
|
||||
"type": "output_text",
|
||||
"text": "1.104.0",
|
||||
"annotations": [
|
||||
{
|
||||
"type": "url_citation",
|
||||
"url": "https://example.com/litellm",
|
||||
"title": "litellm releases",
|
||||
"start_index": 0,
|
||||
"end_index": 7,
|
||||
}
|
||||
],
|
||||
}
|
||||
],
|
||||
},
|
||||
),
|
||||
{"input_tokens": 41, "output_tokens": 5, "total_tokens": 46},
|
||||
)
|
||||
)
|
||||
|
||||
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model=f"openai/{cc.OPENAI_BACKEND}", api_base=wire.url, api_key=cc.OPENAI_API_KEY)
|
||||
response: Final = gateway.request(
|
||||
"POST",
|
||||
"/v1/messages",
|
||||
{**request_body, "model": model},
|
||||
params={"beta": "true"},
|
||||
headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA),
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
payload: Final = cc.JSON_OBJECT.validate_json(response.content)
|
||||
assert payload["content"] == [{"type": "text", "text": "1.104.0"}], payload["content"]
|
||||
assert len(wire.drain()) == 1
|
||||
Loading…
Add table
Reference in a new issue