From 3380361e7164da9f2e7cd7c3975cfa08a91499b1 Mon Sep 17 00:00:00 2001 From: kerry Date: Tue, 29 Sep 2026 22:36:46 +0000 Subject: [PATCH] test(integration): drop the pre-move anthropic test paths Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- ...anthropic_adaptive_thinking_effort_wire.py | 111 ----- .../anthropic/test_anthropic_advisor_wire.py | 201 --------- .../test_anthropic_compaction_wire.py | 108 ----- .../test_anthropic_document_input_wire.py | 133 ------ .../test_anthropic_image_input_wire.py | 67 --- ...ropic_interleaved_thinking_history_wire.py | 167 ------- ...t_anthropic_legacy_thinking_budget_wire.py | 77 ---- .../test_anthropic_long_context_beta_wire.py | 53 --- ..._anthropic_messages_live_lifecycle_wire.py | 82 ---- .../test_anthropic_messages_timeout_wire.py | 58 --- ...est_anthropic_model_switch_history_wire.py | 83 ---- ...est_anthropic_prompt_cache_billing_wire.py | 85 ---- .../test_anthropic_request_fidelity_wire.py | 71 --- ...anthropic_thinking_signature_retry_wire.py | 95 ---- .../test_anthropic_tool_loop_wire.py | 165 ------- ...est_anthropic_web_search_citations_wire.py | 170 -------- .../anthropic/test_anthropic_wire.py | 206 --------- .../test_websearch_interception_wire.py | 409 ------------------ 18 files changed, 2341 deletions(-) delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_adaptive_thinking_effort_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_advisor_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_compaction_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_document_input_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_image_input_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_interleaved_thinking_history_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_legacy_thinking_budget_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_long_context_beta_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_messages_live_lifecycle_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_messages_timeout_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_model_switch_history_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_prompt_cache_billing_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_request_fidelity_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_thinking_signature_retry_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_tool_loop_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_web_search_citations_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_anthropic_wire.py delete mode 100644 tests/integration/messages_endpoint/providers/anthropic/test_websearch_interception_wire.py diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_adaptive_thinking_effort_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_adaptive_thinking_effort_wire.py deleted file mode 100644 index 9f929d69a05..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_adaptive_thinking_effort_wire.py +++ /dev/null @@ -1,111 +0,0 @@ -import uuid -from typing import Final - -import pytest -from integration._support import claude_code as cc -from integration._support.client import Gateway, eventually -from integration._support.database import read_rows -from integration._support.wire import Reply, Request, wire_server -from pydantic import JsonValue - - -def _diff(expected: dict[str, JsonValue], body: dict[str, JsonValue]) -> dict[str, JsonValue]: - return { - key: {"expected": expected.get(key), "upstream": body.get(key)} - for key in expected.keys() | body.keys() - if expected.get(key) != body.get(key) - } - - -def test_adaptive_thinking_and_effort_reach_anthropic_intact(gateway: Gateway) -> None: - identity: Final = f"msg_fable_{uuid.uuid4().hex}" - request_body: Final = cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000) - cli_beta: Final = frozenset(cc.FRONTIER_CLI_BETA.split(",")) - - def respond(request: Request) -> Reply: - assert request.method == "POST" - assert request.target == "/v1/messages", request.target - assert request.headers["x-api-key"] == cc.ANTHROPIC_API_KEY - assert request.headers["anthropic-version"] == "2023-06-01" - upstream_beta: Final = request.headers.get("anthropic-beta", "") - assert cli_beta <= frozenset(upstream_beta.split(",")), upstream_beta - assert upstream_beta.split(",").count("effort-2025-11-24") == 1, upstream_beta - body: Final = cc.JSON_OBJECT.validate_json(request.body) - expected: Final = {**request_body, "model": cc.FABLE} - assert body == expected, _diff(expected, body) - return Reply( - content_type="text/event-stream", - chunks=cc.text_stream(identity, cc.FABLE, "PONG", {"input_tokens": 12, "output_tokens": 4}), - ) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_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 - events: Final = cc.sse_events(response.text) - assert [event for event, _ in events][-1] == "message_stop" - assert events[4][1]["delta"]["stop_reason"] == "end_turn" - assert len(wire.drain()) == 1 - rows: Final = eventually( - lambda: read_rows( - 'SELECT prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', - (identity,), - ), - lambda values: len(values) == 1, - seconds=70, - ) - assert rows[0]["prompt_tokens"] == 12 and rows[0]["completion_tokens"] == 4 - - -def test_xhigh_effort_reaches_anthropic_and_charges_by_usage(gateway: Gateway) -> None: - identity: Final = f"msg_opus_{uuid.uuid4().hex}" - request_body: Final = cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "xhigh", 128000) - cli_beta: Final = frozenset(cc.FRONTIER_CLI_BETA.split(",")) - - def respond(request: Request) -> Reply: - assert request.method == "POST" - assert request.target == "/v1/messages", request.target - assert request.headers["x-api-key"] == cc.ANTHROPIC_API_KEY - upstream_beta: Final = request.headers.get("anthropic-beta", "") - assert cli_beta <= frozenset(upstream_beta.split(",")), upstream_beta - assert upstream_beta.split(",").count("effort-2025-11-24") == 1, upstream_beta - body: Final = cc.JSON_OBJECT.validate_json(request.body) - expected: Final = {**request_body, "model": cc.OPUS} - assert body == expected, _diff(expected, body) - return Reply( - content_type="text/event-stream", - chunks=cc.text_stream(identity, cc.OPUS, "PONG", {"input_tokens": 10, "output_tokens": 5}), - ) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model( - model=f"anthropic/{cc.OPUS}", - api_base=wire.url, - api_key=cc.ANTHROPIC_API_KEY, - input_cost_per_token=1e-6, - output_cost_per_token=5e-6, - ) - 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 - rows: Final = eventually( - lambda: read_rows( - 'SELECT spend, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', - (identity,), - ), - lambda values: len(values) == 1, - seconds=70, - ) - assert float(rows[0]["spend"]) == pytest.approx(10 * 1e-6 + 5 * 5e-6) diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_advisor_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_advisor_wire.py deleted file mode 100644 index b2f44d8f155..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_advisor_wire.py +++ /dev/null @@ -1,201 +0,0 @@ -import json -import uuid -from pathlib import Path -from typing import Final - -import pytest -import yaml -from integration._support.client import Gateway -from integration._support.process import owned_proxy -from integration._support.wire import Reply, Request, wire_server - -_ADVISOR_KEY: Final = "synthetic-advisor-key" -_PROXY_ANTHROPIC_KEY: Final = "sk-proxy-owned-anthropic-secret" -_QUESTION: Final = "which index should this query use" -_ADVICE: Final = "use the composite index on (tenant_id, created_at)" -_FINAL_ANSWER: Final = "done, the composite index is the right one" - - -def _advisor_call_message(question: str) -> dict[str, object]: - return { - "role": "assistant", - "content": None, - "tool_calls": [ - { - "id": "advisor-call", - "type": "function", - "function": {"name": "advisor", "arguments": json.dumps({"question": question})}, - } - ], - } - - -_FINAL_MESSAGE: Final = {"role": "assistant", "content": _FINAL_ANSWER} - - -def _chat_completion(identity: str, message: dict[str, object], finish_reason: str) -> Reply: - return Reply( - body=json.dumps( - { - "id": f"chatcmpl-{identity}", - "object": "chat.completion", - "created": 1, - "model": "llama-3.3-70b-versatile", - "choices": [{"index": 0, "message": message, "finish_reason": finish_reason}], - "usage": {"prompt_tokens": 10, "completion_tokens": 4, "total_tokens": 14}, - } - ).encode() - ) - - -def _executor_reply(body: dict[str, object], identity: str, question: str) -> Reply: - messages: Final = body["messages"] - assert isinstance(messages, list) - if any(message.get("role") == "tool" for message in messages): - assert messages[-1]["content"] == _ADVICE - return _chat_completion(identity, _FINAL_MESSAGE, "stop") - tools: Final = body["tools"] - assert isinstance(tools, list) - assert tools[0]["function"]["name"] == "advisor" - return _chat_completion(identity, _advisor_call_message(question), "tool_calls") - - -@pytest.mark.covers("providers.anthropic_messages_advisor.sub_call_uses_the_configured_advisor_deployment") -def test_advisor_sub_call_reaches_the_router_deployment_with_its_key_instead_of_anthropic_unauthenticated( - gateway: Gateway, -) -> None: - identity: Final = "advisor-wire-" + uuid.uuid4().hex - migration: Final = "please plan the migration " + identity - question: Final = _QUESTION + " " + identity - - def respond(request: Request) -> Reply: - body: Final = json.loads(request.body) - if request.target == "/v1/chat/completions": - assert request.headers["authorization"] == "Bearer integration-provider-key" - return _executor_reply(body, identity, question) - assert request.target == "/v1/messages" - assert request.headers["x-api-key"] == _ADVISOR_KEY - assert body["model"] == "claude-opus-4-1-20250805" - assert body["messages"] == [ - {"role": "user", "content": migration}, - {"role": "user", "content": question}, - ] - assert "tools" not in body - return Reply( - body=json.dumps( - { - "id": f"msg-{identity}", - "type": "message", - "role": "assistant", - "model": "claude-opus-4-1-20250805", - "content": [{"type": "text", "text": _ADVICE}], - "stop_reason": "end_turn", - "stop_sequence": None, - "usage": {"input_tokens": 12, "output_tokens": 6}, - } - ).encode() - ) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - executor: Final = scenario.model(model="hosted_vllm/gpt-4o-mini", api_base=wire.url + "/v1") - advisor: Final = scenario.model( - model="anthropic/claude-opus-4-1-20250805", api_base=wire.url, api_key=_ADVISOR_KEY - ) - response: Final = gateway.request( - "POST", - "/v1/messages", - { - "model": executor, - "max_tokens": 64, - "messages": [{"role": "user", "content": migration}], - "tools": [{"type": "advisor_20260301", "name": "advisor", "model": advisor}], - }, - ) - assert response.status_code == 200, response.text - body: Final = response.json() - assert body["content"] == [{"type": "text", "text": _FINAL_ANSWER}], response.text - assert body["stop_reason"] == "end_turn", response.text - assert [request.target for request in wire.drain()] == [ - "/v1/chat/completions", - "/v1/messages", - "/v1/chat/completions", - ] - - -def _advice_reply(identity: str) -> Reply: - return Reply( - body=json.dumps( - { - "id": f"msg-{identity}", - "type": "message", - "role": "assistant", - "model": "claude-opus-4-1-20250805", - "content": [{"type": "text", "text": _ADVICE}], - "stop_reason": "end_turn", - "stop_sequence": None, - "usage": {"input_tokens": 12, "output_tokens": 6}, - } - ).encode() - ) - - -@pytest.mark.covers("providers.anthropic_messages_advisor.caller_api_base_without_api_key_never_receives_the_proxy_key") -def test_advisor_api_base_without_api_key_is_rejected_before_the_proxy_anthropic_key_reaches_the_caller_host( - gateway: Gateway, tmp_path: Path -) -> None: - identity: Final = "advisor-leak-" + uuid.uuid4().hex - question: Final = _QUESTION + " " + identity - - def executor(request: Request) -> Reply: - assert request.target == "/v1/chat/completions", request.target - return _executor_reply(json.loads(request.body), identity, question) - - def caller_host(request: Request) -> Reply: - return _advice_reply(identity) - - config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) - config["general_settings"]["allow_client_side_credentials"] = True - path: Final = tmp_path / "client-side-credentials.yaml" - path.write_text(yaml.safe_dump(config)) - with ( - wire_server(executor) as executor_wire, - wire_server(caller_host) as caller_wire, - owned_proxy(gateway, tmp_path, {"ANTHROPIC_API_KEY": _PROXY_ANTHROPIC_KEY}, config=path) as candidate, - candidate.scenario() as scenario, - ): - model: Final = scenario.model(model="hosted_vllm/gpt-4o-mini", api_base=executor_wire.url + "/v1") - response: Final = candidate.request( - "POST", - "/v1/messages", - { - "model": model, - "max_tokens": 64, - "messages": [{"role": "user", "content": "please plan the migration"}], - "tools": [ - { - "type": "advisor_20260301", - "name": "advisor", - "model": "anthropic/claude-opus-4-1-20250805", - "api_base": caller_wire.url, - } - ], - }, - ) - received: Final = caller_wire.drain() - assert [ - (request.target, request.headers.get("x-api-key"), json.loads(request.body)["messages"]) - for request in received - ] == [], response.text - assert response.is_error, response.text - assert response.json() == { - "type": "error", - "error": { - "type": "api_error", - "message": ( - "advisor tool definition sets 'api_base' without 'api_key'. A caller-supplied api_base is only " - "honored alongside a caller-supplied api_key, so the proxy's own credentials are never sent to a " - "caller-chosen destination." - ), - }, - }, response.text - assert executor_wire.drain() == (), response.text diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_compaction_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_compaction_wire.py deleted file mode 100644 index c6ac00c20e8..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_compaction_wire.py +++ /dev/null @@ -1,108 +0,0 @@ -import uuid -from typing import Final - -from integration._support import claude_code as cc -from integration._support.client import Gateway -from integration._support.wire import Reply, Request, wire_server - -_CONTEXT_MANAGEMENT: Final = { - "edits": [ - {"type": "clear_thinking_20251015", "keep": "all"}, - {"type": "compact_20260112", "trigger": {"type": "input_tokens", "value": 150000}}, - ] -} -_COMPACTION_BLOCK: Final = {"type": "compaction", "content": ""} - - -def _compaction_stream(identity: str) -> tuple[bytes, ...]: - return ( - cc.sse_frame( - "message_start", - { - "type": "message_start", - "message": { - "id": identity, - "type": "message", - "role": "assistant", - "model": cc.FABLE, - "content": [], - "stop_reason": None, - "stop_sequence": None, - "usage": {"input_tokens": 20, "output_tokens": 1}, - }, - }, - ), - cc.sse_frame( - "content_block_start", - {"type": "content_block_start", "index": 0, "content_block": dict(_COMPACTION_BLOCK)}, - ), - cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}), - cc.sse_frame( - "message_delta", - { - "type": "message_delta", - "delta": {"stop_reason": "end_turn", "stop_sequence": None}, - "usage": {"output_tokens": 5}, - "context_management": { - "applied_edits": [{"type": "compact_20260112", "compacted_at": "2026-09-26T00:00:00Z"}] - }, - }, - ), - cc.sse_frame("message_stop", {"type": "message_stop"}), - ) - - -def test_compaction_edit_and_applied_edit_block_round_trip_through_anthropic(gateway: Gateway) -> None: - identity: Final = f"msg_cm_{uuid.uuid4().hex}" - request_body: Final = { - **cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000), - "context_management": _CONTEXT_MANAGEMENT, - } - turn3: Final = cc.tool_loop_turn2( - request_body, - ( - dict(_COMPACTION_BLOCK), - {"type": "text", "text": "continuing after compaction"}, - ), - (), - ) - first_expected: Final = {**request_body, "model": cc.FABLE} - second_expected: Final = {**turn3, "model": cc.FABLE} - - def respond(request: Request) -> Reply: - assert request.method == "POST" - assert request.target == "/v1/messages", request.target - body: Final = cc.JSON_OBJECT.validate_json(request.body) - if body == first_expected: - assert body["context_management"] == _CONTEXT_MANAGEMENT - return Reply(content_type="text/event-stream", chunks=_compaction_stream(identity)) - assert body == second_expected, { - key: (second_expected.get(key), body.get(key)) - for key in second_expected.keys() | body.keys() - if second_expected.get(key) != body.get(key) - } - assert body["context_management"] == _CONTEXT_MANAGEMENT - return Reply( - content_type="text/event-stream", - chunks=cc.text_stream("msg_cm_next", cc.FABLE, "OK", {"input_tokens": 20, "output_tokens": 2}), - ) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) - headers: Final = cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA) - response1: Final = gateway.request( - "POST", "/v1/messages", {**request_body, "model": model}, params={"beta": "true"}, headers=headers - ) - assert response1.status_code == 200, response1.text - events: Final = cc.sse_events(response1.text) - assert events[1][1]["content_block"] == _COMPACTION_BLOCK, events[1] - deltas: Final = [data for event, data in events if event == "message_delta"] - assert len(deltas) == 1 and deltas[0].get("context_management") == { - "applied_edits": [{"type": "compact_20260112", "compacted_at": "2026-09-26T00:00:00Z"}] - }, events - response2: Final = gateway.request( - "POST", "/v1/messages", {**turn3, "model": model}, params={"beta": "true"}, headers=headers - ) - assert response2.status_code == 200, response2.text - bodies: Final = tuple(cc.JSON_OBJECT.validate_json(request.body) for request in wire.drain()) - assert bodies == (first_expected, second_expected), bodies diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_document_input_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_document_input_wire.py deleted file mode 100644 index 48b914a7374..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_document_input_wire.py +++ /dev/null @@ -1,133 +0,0 @@ -import base64 -import uuid -from typing import Final - -from integration._support import claude_code as cc -from integration._support.client import Gateway -from integration._support.wire import Reply, Request, wire_server - -_PDF_BYTES: Final = ( - b"%PDF-1.1\n" - b"1 0 obj<>endobj\n" - b"2 0 obj<>endobj\n" - b"3 0 obj<>endobj\n" - b"trailer<>\n%%EOF" -) -_DOC_BLOCK: Final = { - "type": "document", - "source": {"type": "base64", "data": base64.b64encode(_PDF_BYTES).decode(), "media_type": "application/pdf"}, - "citations": {"enabled": True}, -} - - -def _cited_stream(identity: str) -> tuple[bytes, ...]: - return ( - cc.sse_frame( - "message_start", - { - "type": "message_start", - "message": { - "id": identity, - "type": "message", - "role": "assistant", - "model": cc.FABLE, - "content": [], - "stop_reason": None, - "stop_sequence": None, - "usage": {"input_tokens": 20, "output_tokens": 1}, - }, - }, - ), - cc.sse_frame( - "content_block_start", - {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, - ), - cc.sse_frame( - "content_block_delta", - {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "A page."}}, - ), - cc.sse_frame( - "content_block_delta", - { - "type": "content_block_delta", - "index": 0, - "delta": { - "type": "citations_delta", - "citation": { - "type": "page_location", - "document_index": 0, - "document_title": "dot.pdf", - "start_page_number": 1, - "end_page_number": 1, - "cited_text": "Page", - }, - }, - }, - ), - cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}), - cc.sse_frame( - "message_delta", - { - "type": "message_delta", - "delta": {"stop_reason": "end_turn", "stop_sequence": None}, - "usage": {"output_tokens": 6}, - }, - ), - cc.sse_frame("message_stop", {"type": "message_stop"}), - ) - - -def test_base64_pdf_document_with_citations_reaches_anthropic_identical(gateway: Gateway) -> None: - request_body: Final = { - **cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000), - "messages": [ - { - "role": "user", - "content": [ - dict(_DOC_BLOCK), - {"type": "text", "text": f"What is on page one? {uuid.uuid4().hex}"}, - ], - } - ], - } - - def respond(request: Request) -> Reply: - assert request.method == "POST" - assert request.target == "/v1/messages", request.target - body: Final = cc.JSON_OBJECT.validate_json(request.body) - expected: Final = {**request_body, "model": cc.FABLE} - assert body == expected, { - key: (expected.get(key), 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=_cited_stream(f"msg_doc_{uuid.uuid4().hex}")) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_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 - events: Final = cc.sse_events(response.text) - citations: Final = [ - data["delta"] for event, data in events if data.get("delta", {}).get("type") == "citations_delta" - ] - assert citations == [ - { - "type": "citations_delta", - "citation": { - "type": "page_location", - "document_index": 0, - "document_title": "dot.pdf", - "start_page_number": 1, - "end_page_number": 1, - "cited_text": "Page", - }, - } - ], citations - assert len(wire.drain()) == 1 diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_image_input_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_image_input_wire.py deleted file mode 100644 index 8a929ee2a06..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_image_input_wire.py +++ /dev/null @@ -1,67 +0,0 @@ -import uuid -from typing import Final - -from integration._support import claude_code as cc -from integration._support.client import Gateway -from integration._support.wire import Reply, Request, wire_server - -_PNG_B64: Final = "iVBORw0KGgoAAAANSUhEUgAAAAQAAAAECAIAAAAmkwkpAAAAEElEQVR4nGP4z8AARwzEcQCukw/x0F8jngAAAABJRU5ErkJggg==" -_IMAGE_BLOCK: Final = { - "type": "image", - "source": {"type": "base64", "data": _PNG_B64, "media_type": "image/png"}, -} - - -def test_tool_result_image_block_and_pasted_image_reach_anthropic_identical(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", - ) - with_image_result: Final = cc.tool_loop_turn2( - turn1, - ({"type": "tool_use", "id": "toolu_img", "name": "Read", "input": {"file_path": "/tmp/cc_probe/dot.png"}},), - (("toolu_img", [dict(_IMAGE_BLOCK)]),), - ) - pasted: Final = { - **cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000), - "messages": [ - { - "role": "user", - "content": [ - dict(_IMAGE_BLOCK), - {"type": "text", "text": f"What colour is this? {uuid.uuid4().hex}"}, - ], - } - ], - } - first_expected: Final = {**with_image_result, "model": cc.FABLE} - second_expected: Final = {**pasted, "model": cc.FABLE} - - def respond(request: Request) -> Reply: - assert request.method == "POST" - assert request.target == "/v1/messages", request.target - body: Final = cc.JSON_OBJECT.validate_json(request.body) - if body != first_expected: - assert body == second_expected, body - return Reply( - content_type="text/event-stream", - chunks=cc.text_stream( - f"msg_img_{uuid.uuid4().hex}", cc.FABLE, "RED", {"input_tokens": 20, "output_tokens": 2} - ), - ) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) - headers: Final = cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA) - response1: Final = gateway.request( - "POST", "/v1/messages", {**with_image_result, "model": model}, params={"beta": "true"}, headers=headers - ) - assert response1.status_code == 200, response1.text - response2: Final = gateway.request( - "POST", "/v1/messages", {**pasted, "model": model}, params={"beta": "true"}, headers=headers - ) - assert response2.status_code == 200, response2.text - bodies: Final = tuple(cc.JSON_OBJECT.validate_json(request.body) for request in wire.drain()) - assert bodies == (first_expected, second_expected), bodies diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_interleaved_thinking_history_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_interleaved_thinking_history_wire.py deleted file mode 100644 index 6990f33e51c..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_interleaved_thinking_history_wire.py +++ /dev/null @@ -1,167 +0,0 @@ -import uuid -from typing import Final - -from integration._support import claude_code as cc -from integration._support.client import Gateway -from integration._support.wire import Reply, Request, wire_server -from pydantic import JsonValue - -_TURN1_TOOL: Final = ("toolu_a", "Read", {"file_path": "/tmp/cc_probe/a.txt"}) -_TURN2_TOOL: Final = ("toolu_b", "Read", {"file_path": "/tmp/cc_probe/b.txt"}) - - -def _diff(expected: dict[str, JsonValue], body: dict[str, JsonValue]) -> dict[str, JsonValue]: - return { - key: {"expected": expected.get(key), "upstream": body.get(key)} - for key in expected.keys() | body.keys() - if expected.get(key) != body.get(key) - } - - -def _turn2(base: dict[str, JsonValue]) -> dict[str, JsonValue]: - return cc.tool_loop_turn2( - base, - ( - {"type": "thinking", "thinking": "plan", "signature": "sig1"}, - {"type": "tool_use", "id": _TURN1_TOOL[0], "name": _TURN1_TOOL[1], "input": _TURN1_TOOL[2]}, - ), - ((_TURN1_TOOL[0], "ALPHA"),), - ) - - -def _turn3(turn2: dict[str, JsonValue]) -> dict[str, JsonValue]: - return cc.tool_loop_turn2( - turn2, - ( - {"type": "thinking", "thinking": "got A", "signature": "sig2"}, - {"type": "text", "text": "got A"}, - {"type": "tool_use", "id": _TURN2_TOOL[0], "name": _TURN2_TOOL[1], "input": _TURN2_TOOL[2]}, - ), - ((_TURN2_TOOL[0], "BRAVO"),), - ) - - -def _interleaved_stream(identity: str) -> tuple[bytes, ...]: - usage: Final = {"input_tokens": 20, "output_tokens": 12} - return ( - cc.sse_frame( - "message_start", - { - "type": "message_start", - "message": { - "id": identity, - "type": "message", - "role": "assistant", - "model": cc.FABLE, - "content": [], - "stop_reason": None, - "stop_sequence": None, - "usage": {"input_tokens": usage["input_tokens"], "output_tokens": 1}, - }, - }, - ), - cc.sse_frame( - "content_block_start", - {"type": "content_block_start", "index": 0, "content_block": {"type": "thinking", "thinking": ""}}, - ), - cc.sse_frame( - "content_block_delta", - {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "got A"}}, - ), - cc.sse_frame( - "content_block_delta", - {"type": "content_block_delta", "index": 0, "delta": {"type": "signature_delta", "signature": "sig2"}}, - ), - cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}), - cc.sse_frame( - "content_block_start", - {"type": "content_block_start", "index": 1, "content_block": {"type": "text", "text": ""}}, - ), - cc.sse_frame( - "content_block_delta", - {"type": "content_block_delta", "index": 1, "delta": {"type": "text_delta", "text": "got A"}}, - ), - cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 1}), - cc.sse_frame( - "content_block_start", - { - "type": "content_block_start", - "index": 2, - "content_block": {"type": "tool_use", "id": _TURN2_TOOL[0], "name": _TURN2_TOOL[1], "input": {}}, - }, - ), - cc.sse_frame( - "content_block_delta", - { - "type": "content_block_delta", - "index": 2, - "delta": {"type": "input_json_delta", "partial_json": '{"file_path": "/tmp/cc_pr'}, - }, - ), - cc.sse_frame( - "content_block_delta", - { - "type": "content_block_delta", - "index": 2, - "delta": {"type": "input_json_delta", "partial_json": 'obe/b.txt"}'}, - }, - ), - cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 2}), - cc.sse_frame( - "message_delta", - { - "type": "message_delta", - "delta": {"stop_reason": "tool_use", "stop_sequence": None}, - "usage": {"output_tokens": usage["output_tokens"]}, - }, - ), - cc.sse_frame("message_stop", {"type": "message_stop"}), - ) - - -def test_interleaved_thinking_text_and_tool_use_history_reaches_anthropic_identical(gateway: Gateway) -> None: - identity: Final = f"msg_il_{uuid.uuid4().hex}" - 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 and reply with both words", - ) - turn2: Final = _turn2(turn1) - turn3: Final = _turn3(turn2) - turn2_expected: Final = {**turn2, "model": cc.FABLE} - turn3_expected: Final = {**turn3, "model": cc.FABLE} - - def respond(request: Request) -> Reply: - assert request.method == "POST" - assert request.target == "/v1/messages", request.target - upstream_beta: Final = request.headers.get("anthropic-beta", "") - assert upstream_beta.split(",").count("interleaved-thinking-2025-05-14") == 1, upstream_beta - body: Final = cc.JSON_OBJECT.validate_json(request.body) - if body == turn2_expected: - return Reply( - content_type="text/event-stream", - chunks=cc.text_stream("msg_il_turn2", cc.FABLE, "got A", {"input_tokens": 20, "output_tokens": 4}), - ) - assert body == turn3_expected, _diff(turn3_expected, body) - return Reply(content_type="text/event-stream", chunks=_interleaved_stream(identity)) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) - headers: Final = cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA) - response2: Final = gateway.request( - "POST", "/v1/messages", {**turn2, "model": model}, params={"beta": "true"}, headers=headers - ) - assert response2.status_code == 200, response2.text - response3: Final = gateway.request( - "POST", "/v1/messages", {**turn3, "model": model}, params={"beta": "true"}, headers=headers - ) - assert response3.status_code == 200, response3.text - events: Final = cc.sse_events(response3.text) - started: Final = [ - (data["index"], data["content_block"]["type"]) for event, data in events if event == "content_block_start" - ] - assert started == [(0, "thinking"), (1, "text"), (2, "tool_use")], started - assert events[-1][0] == "message_stop" - bodies: Final = tuple(cc.JSON_OBJECT.validate_json(request.body) for request in wire.drain()) - assert bodies == (turn2_expected, turn3_expected), bodies diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_legacy_thinking_budget_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_legacy_thinking_budget_wire.py deleted file mode 100644 index 242e5c7ec5a..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_legacy_thinking_budget_wire.py +++ /dev/null @@ -1,77 +0,0 @@ -import json -import uuid -from typing import Final - -import pytest -from integration._support.client import Gateway -from integration._support.wire import Reply, Request, wire_server - -_MODEL: Final = "claude-sonnet-4-6" -_KEY: Final = "synthetic-anthropic-key" -_THINKING: Final = {"type": "enabled", "budget_tokens": 8000} -_TOOL: Final = { - "name": "read_file", - "description": "read a file", - "input_schema": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}, -} -_NEXT_CALL: Final = {"type": "tool_use", "id": "call-2", "name": "read_file", "input": {"path": "schema.prisma"}} - - -def _tool_loop_history(identity: str) -> tuple[dict[str, object], ...]: - return ( - {"role": "user", "content": f"open the config for {identity}"}, - { - "role": "assistant", - "content": [{"type": "tool_use", "id": "call-1", "name": "read_file", "input": {"path": "config.yaml"}}], - }, - {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "call-1", "content": "model_list: []"}]}, - ) - - -def _tool_use_reply(identity: str) -> Reply: - return Reply( - body=json.dumps( - { - "id": f"msg-{identity}", - "type": "message", - "role": "assistant", - "model": _MODEL, - "content": [_NEXT_CALL], - "stop_reason": "tool_use", - "stop_sequence": None, - "usage": {"input_tokens": 40, "output_tokens": 12}, - } - ).encode() - ) - - -@pytest.mark.covers("providers.anthropic_messages.claude_4_6_legacy_thinking_budget_reaches_the_wire_unchanged") -def test_claude_4_6_thinking_budget_tokens_on_messages_is_forwarded_instead_of_rewritten_to_adaptive( - gateway: Gateway, -) -> None: - identity: Final = "legacy-thinking-" + uuid.uuid4().hex - history: Final = _tool_loop_history(identity) - - def respond(request: Request) -> Reply: - assert request.method == "POST" and request.target == "/v1/messages", request.target - assert request.headers["x-api-key"] == _KEY - body: Final = json.loads(request.body) - assert body["thinking"] == _THINKING, body - assert "output_config" not in body, body - assert body["max_tokens"] == 32768, body - assert body["messages"] == list(history), body - assert body["tools"] == [_TOOL], body - return _tool_use_reply(identity) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model(model=f"anthropic/{_MODEL}", api_base=wire.url, api_key=_KEY) - response: Final = gateway.request( - "POST", - "/v1/messages", - {"model": model, "max_tokens": 32768, "thinking": _THINKING, "messages": history, "tools": [_TOOL]}, - ) - assert response.status_code == 200, response.text - body: Final = response.json() - assert body["content"] == [_NEXT_CALL], response.text - assert body["stop_reason"] == "tool_use", response.text - assert len(wire.drain()) == 1 diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_long_context_beta_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_long_context_beta_wire.py deleted file mode 100644 index 7fe405ca439..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_long_context_beta_wire.py +++ /dev/null @@ -1,53 +0,0 @@ -import uuid -from typing import Final - -import pytest -from integration._support import claude_code as cc -from integration._support.client import Gateway, eventually -from integration._support.database import read_rows -from integration._support.wire import Reply, Request, wire_server - -_BETA_1M: Final = f"{cc.FRONTIER_CLI_BETA.replace(',effort-2025-11-24', ',context-1m-2025-08-07,effort-2025-11-24')}" - - -def test_1m_context_beta_forwarded_and_tiered_prompt_priced_above_200k(gateway: Gateway) -> None: - identity: Final = f"msg_1m_{uuid.uuid4().hex}" - 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 == "/v1/messages", request.target - upstream_beta: Final = request.headers.get("anthropic-beta", "") - assert upstream_beta.split(",").count("context-1m-2025-08-07") == 1, upstream_beta - return Reply( - content_type="text/event-stream", - chunks=cc.text_stream(identity, cc.FABLE, "PONG", {"input_tokens": 250000, "output_tokens": 100}), - ) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model( - model=f"anthropic/{cc.FABLE}", - api_base=wire.url, - api_key=cc.ANTHROPIC_API_KEY, - input_cost_per_token=1e-6, - input_cost_per_token_above_200k_tokens=2e-6, - output_cost_per_token=5e-6, - ) - response: Final = gateway.request( - "POST", - "/v1/messages", - {**request_body, "model": model}, - params={"beta": "true"}, - headers=cc.cli_headers(gateway.key, _BETA_1M), - ) - assert response.status_code == 200, response.text - assert len(wire.drain()) == 1 - rows: Final = eventually( - lambda: read_rows( - 'SELECT spend, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', - (identity,), - ), - lambda values: len(values) == 1, - seconds=70, - ) - assert float(rows[0]["spend"]) == pytest.approx(250000 * 2e-6 + 100 * 5e-6), dict(rows[0]) diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_messages_live_lifecycle_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_messages_live_lifecycle_wire.py deleted file mode 100644 index cb7043c0362..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_messages_live_lifecycle_wire.py +++ /dev/null @@ -1,82 +0,0 @@ -import json -import threading -import uuid -from typing import Final - -from integration._support.client import Gateway -from integration._support.wire import Reply, Request, wire_server - -_MODEL: Final = "claude-sonnet-4-5-20250929" -_API_KEY: Final = "synthetic-anthropic-key" - - -def _sse(event: str, payload: dict[str, object]) -> bytes: - return f"event: {event}\ndata: {json.dumps(payload)}\n\n".encode() - - -def test_messages_stream_message_start_reaches_client_before_content_without_fallback( - gateway: Gateway, -) -> None: - """With no fallback able to take over, the proxy must not hold lifecycle - frames back for a retry that cannot happen: message_start reaches the - client while the upstream is still thinking.""" - gate: Final = threading.Event() - head: Final = _sse("message_start", {"type": "message_start", "message": {"id": "msg_live_1"}}) + _sse( - "content_block_start", - {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, - ) - tail: Final = ( - _sse( - "content_block_delta", - {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Hello"}}, - ) - + _sse("content_block_stop", {"type": "content_block_stop", "index": 0}) - + _sse( - "message_delta", - {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"output_tokens": 3}}, - ) - + _sse("message_stop", {"type": "message_stop"}) - ) - prompt: Final = "live-lifecycle-" + uuid.uuid4().hex - - def respond(request: Request) -> Reply: - assert request.method == "POST" and request.target == "/v1/messages" - assert request.headers["x-api-key"] == _API_KEY - body: Final = json.loads(request.body) - assert body["model"] == _MODEL - assert body["stream"] is True - assert body["messages"] == [{"role": "user", "content": prompt}] - return Reply(content_type="text/event-stream", chunks=(head, tail), gate_after_first=gate) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model(model=f"anthropic/{_MODEL}", api_base=wire.url, api_key=_API_KEY) - with gateway.client.stream( - "POST", - "/v1/messages", - json={ - "model": model, - "max_tokens": 16, - "stream": True, - "messages": [{"role": "user", "content": prompt}], - }, - headers={"Authorization": f"Bearer {gateway.key}"}, - ) as response: - assert response.status_code == 200, response.read().decode() - lines = response.iter_lines() - first_event: Final = next( - json.loads(line.removeprefix("data: ")) for line in lines if line.startswith("data: ") - ) - assert first_event["type"] == "message_start" - gate.set() - events: Final = (first_event,) + tuple( - json.loads(line.removeprefix("data: ")) for line in lines if line.startswith("data: ") - ) - assert tuple(event["type"] for event in events) == ( - "message_start", - "content_block_start", - "content_block_delta", - "content_block_stop", - "message_delta", - "message_stop", - ), f"observed events: {events!r}" - assert [request.target for request in wire.drain()] == ["/v1/messages"] diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_messages_timeout_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_messages_timeout_wire.py deleted file mode 100644 index 76c29ad7763..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_messages_timeout_wire.py +++ /dev/null @@ -1,58 +0,0 @@ -import json -import time -import uuid -from typing import Final - -import pytest -from integration._support.client import JSON_OBJECT, Gateway -from integration._support.wire import Reply, Request, wire_server - -_UPSTREAM_STALL_SECONDS: Final = 4.0 -_CONFIGURED_TIMEOUT_SECONDS: Final = 1.0 - - -@pytest.mark.covers("providers.anthropic_messages.configured_timeout_aborts_stalled_upstream") -def test_messages_endpoint_honors_configured_timeout_against_stalled_upstream(gateway: Gateway) -> None: - prompt: Final = "stall-" + uuid.uuid4().hex - - def respond(request: Request) -> Reply: - assert request.method == "POST" and request.target == "/v1/messages" - assert request.headers["x-api-key"] == "synthetic-anthropic-key" - body: Final = JSON_OBJECT.validate_json(request.body) - assert body["model"] == "claude-sonnet-4-5-20250929" - assert body["messages"] == [{"role": "user", "content": prompt}] - assert body["max_tokens"] == 16 - assert "timeout" not in body - time.sleep(_UPSTREAM_STALL_SECONDS) - return Reply( - body=json.dumps( - { - "id": "msg_stalled", - "type": "message", - "role": "assistant", - "model": "claude-sonnet-4-5-20250929", - "content": [{"type": "text", "text": "too late"}], - "stop_reason": "end_turn", - "stop_sequence": None, - "usage": {"input_tokens": 1, "output_tokens": 2}, - } - ).encode() - ) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model( - model="anthropic/claude-sonnet-4-5-20250929", - api_base=wire.url, - api_key="synthetic-anthropic-key", - timeout=_CONFIGURED_TIMEOUT_SECONDS, - ) - started: Final = time.monotonic() - response: Final = gateway.request( - "POST", - "/v1/messages", - {"model": model, "max_tokens": 16, "messages": [{"role": "user", "content": prompt}]}, - ) - elapsed: Final = time.monotonic() - started - assert response.status_code == 408, response.text - assert elapsed < _UPSTREAM_STALL_SECONDS, f"timed out only after {elapsed:.2f}s: {response.text}" - assert len(wire.drain()) == 1 diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_model_switch_history_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_model_switch_history_wire.py deleted file mode 100644 index 5753d395bc4..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_model_switch_history_wire.py +++ /dev/null @@ -1,83 +0,0 @@ -import uuid -from typing import Final - -from integration._support import claude_code as cc -from integration._support.client import Gateway, eventually -from integration._support.database import read_rows -from integration._support.wire import Reply, Request, wire_server - - -def test_mid_loop_model_switch_replays_history_byte_identical(gateway: Gateway) -> None: - identity1: Final = f"msg_sw1_{uuid.uuid4().hex}" - identity2: Final = f"msg_sw2_{uuid.uuid4().hex}" - 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"),), - ) - first_expected: Final = {**turn1, "model": cc.FABLE} - second_expected: Final = {**turn2, "model": cc.OPUS} - - def respond(request: Request) -> Reply: - assert request.method == "POST" - assert request.target == "/v1/messages", request.target - body: Final = cc.JSON_OBJECT.validate_json(request.body) - if body == first_expected: - return Reply( - content_type="text/event-stream", - chunks=cc.tool_use_stream( - identity1, - 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}, - ), - ) - assert body == second_expected, { - key: (second_expected.get(key), body.get(key)) - for key in second_expected.keys() | body.keys() - if second_expected.get(key) != body.get(key) - } - return Reply( - content_type="text/event-stream", - chunks=cc.text_stream(identity2, cc.OPUS, "PROBE", {"input_tokens": 30, "output_tokens": 3}), - ) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - fable: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) - opus: Final = scenario.model(model=f"anthropic/{cc.OPUS}", api_base=wire.url, api_key=cc.ANTHROPIC_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": opus}, params={"beta": "true"}, headers=headers - ) - assert response2.status_code == 200, response2.text - bodies: Final = tuple(cc.JSON_OBJECT.validate_json(request.body) for request in wire.drain()) - assert bodies == (first_expected, second_expected), bodies - rows: Final = eventually( - lambda: read_rows( - 'SELECT model FROM "LiteLLM_SpendLogs" WHERE request_id=%s', - (identity2,), - ), - lambda values: len(values) == 1, - seconds=70, - ) - assert rows[0]["model"] == f"anthropic/{cc.OPUS}" diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_prompt_cache_billing_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_prompt_cache_billing_wire.py deleted file mode 100644 index 06318b52aee..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_prompt_cache_billing_wire.py +++ /dev/null @@ -1,85 +0,0 @@ -import uuid -from typing import Final - -import pytest -from integration._support import claude_code as cc -from integration._support.client import Gateway, eventually -from integration._support.database import read_rows -from integration._support.wire import Reply, Request, wire_server - -_USAGE: Final = { - "input_tokens": 10, - "cache_read_input_tokens": 3000, - "cache_creation_input_tokens": 200, - "output_tokens": 5, -} - - -def test_cached_turn_charges_cache_read_and_creation_rates(gateway: Gateway) -> None: - identity: Final = f"msg_pc_{uuid.uuid4().hex}" - 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"),), - ) - - def respond(request: Request) -> Reply: - assert request.method == "POST" - assert request.target == "/v1/messages", request.target - body: Final = cc.JSON_OBJECT.validate_json(request.body) - expected: Final = {**turn2, "model": cc.FABLE} - assert body == expected, { - key: (expected.get(key), 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.text_stream(identity, cc.FABLE, "PROBE", _USAGE)) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model( - model=f"anthropic/{cc.FABLE}", - api_base=wire.url, - api_key=cc.ANTHROPIC_API_KEY, - input_cost_per_token=1e-6, - output_cost_per_token=5e-6, - cache_read_input_token_cost=1e-7, - cache_creation_input_token_cost=1.25e-6, - ) - 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 - events: Final = cc.sse_events(response.text) - usage: Final = events[0][1]["message"]["usage"] - assert usage["cache_read_input_tokens"] == 3000, usage - assert usage["cache_creation_input_tokens"] == 200, usage - assert len(wire.drain()) == 1 - rows: Final = eventually( - lambda: read_rows( - 'SELECT spend, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', - (identity,), - ), - lambda values: len(values) == 1, - seconds=70, - ) - assert float(rows[0]["spend"]) == pytest.approx(10 * 1e-6 + 3000 * 1e-7 + 200 * 1.25e-6 + 5 * 5e-6), dict( - rows[0] - ) diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_request_fidelity_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_request_fidelity_wire.py deleted file mode 100644 index a5a57737fc1..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_request_fidelity_wire.py +++ /dev/null @@ -1,71 +0,0 @@ -import uuid -from typing import Final - -from integration._support import claude_code as cc -from integration._support.client import Gateway, eventually -from integration._support.database import read_rows -from integration._support.wire import Reply, Request, wire_server - -_MODEL: Final = cc.SONNET - - -def test_streaming_request_reaches_anthropic_intact_and_streams_back(gateway: Gateway) -> None: - identity: Final = f"msg_cc_{uuid.uuid4().hex}" - request_body: Final = cc.claude_code_request(f"cache-bust-{uuid.uuid4().hex}") - cli_beta: Final = frozenset(cc.CLI_BETA.split(",")) - - def respond(request: Request) -> Reply: - assert request.method == "POST" - assert request.target == "/v1/messages", request.target - assert request.headers["x-api-key"] == cc.ANTHROPIC_API_KEY - assert request.headers["anthropic-version"] == "2023-06-01" - assert frozenset(request.headers.get("anthropic-beta", "").split(",")) == cli_beta, request.headers.get( - "anthropic-beta" - ) - assert "authorization" not in request.headers, dict(request.headers) - assert all(gateway.key not in value for value in request.headers.values()), dict(request.headers) - body: Final = cc.JSON_OBJECT.validate_json(request.body) - expected: Final = {**request_body, "model": _MODEL} - assert body == expected, { - key: (expected.get(key), 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.text_stream(identity, _MODEL, "PONG", {"input_tokens": 12, "output_tokens": 4}), - ) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model(model=f"anthropic/{_MODEL}", api_base=wire.url, api_key=cc.ANTHROPIC_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 response.headers["content-type"].startswith("text/event-stream"), dict(response.headers) - events: Final = cc.sse_events(response.text) - assert [event for event, _ in events] == [ - "message_start", - "content_block_start", - "content_block_delta", - "content_block_stop", - "message_delta", - "message_stop", - ] - assert events[2][1]["delta"] == {"type": "text_delta", "text": "PONG"} - assert events[4][1]["delta"]["stop_reason"] == "end_turn" - assert events[4][1]["usage"]["output_tokens"] == 4 - assert len(wire.drain()) == 1 - rows: Final = eventually( - lambda: read_rows( - 'SELECT prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', - (identity,), - ), - lambda values: len(values) == 1, - seconds=70, - ) - assert rows[0]["prompt_tokens"] == 12 and rows[0]["completion_tokens"] == 4 diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_thinking_signature_retry_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_thinking_signature_retry_wire.py deleted file mode 100644 index e414d8f0d11..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_thinking_signature_retry_wire.py +++ /dev/null @@ -1,95 +0,0 @@ -import json -import uuid -from typing import Final - -import pytest -from integration._support.client import Gateway -from integration._support.wire import Reply, Request, wire_server - -MODEL: Final = "claude-sonnet-4-5-20250929" -KEY: Final = "synthetic-anthropic-key" -SIGNATURE_ERROR: Final = json.dumps( - { - "type": "error", - "error": { - "type": "invalid_request_error", - "message": "messages.2.content.0.thinking.signature.str: Input should be a valid string", - }, - } -).encode() -TOOLS: Final = ({"name": "lookup", "input_schema": {"type": "object", "properties": {"key": {"type": "string"}}}},) - - -def _history_with_unsigned_thinking(identity: str) -> tuple[dict[str, object], ...]: - return ( - {"role": "user", "content": [{"type": "text", "text": f"first question {identity}"}]}, - {"role": "assistant", "content": [{"type": "text", "text": "first answer"}]}, - {"role": "user", "content": [{"type": "text", "text": "second question"}]}, - { - "role": "assistant", - "content": [ - {"type": "thinking", "thinking": "replayed from another provider", "signature": None}, - {"type": "tool_use", "id": "call-1", "name": "lookup", "input": {"key": "value"}}, - ], - }, - {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "call-1", "content": "found"}]}, - ) - - -@pytest.mark.covers("providers.anthropic_messages.missing_thinking_signature_400_retries_without_thinking_blocks") -def test_missing_thinking_signature_400_retries_once_without_thinking_blocks_and_returns_200( - gateway: Gateway, -) -> None: - identity: Final = "thinking-signature-" + uuid.uuid4().hex - history: Final = _history_with_unsigned_thinking(identity) - tool_use_only_turn: Final = { - "role": "assistant", - "content": [{"type": "tool_use", "id": "call-1", "name": "lookup", "input": {"key": "value"}}], - } - - def respond(request: Request) -> Reply: - assert request.method == "POST" and request.target == "/v1/messages" - assert request.headers["x-api-key"] == KEY - body: Final = json.loads(request.body) - assert body["model"] == MODEL - assert body["tools"] == list(TOOLS), body - if body["messages"][3]["content"][0]["type"] == "thinking": - assert body["messages"] == list(history), body - assert body["thinking"] == {"type": "enabled", "budget_tokens": 1024}, body - return Reply(status=400, body=SIGNATURE_ERROR) - assert body["messages"] == [*history[:3], tool_use_only_turn, history[4]], body - assert "thinking" not in body, body - return Reply( - body=json.dumps( - { - "id": identity, - "type": "message", - "role": "assistant", - "model": MODEL, - "content": [{"type": "text", "text": "recovered without thinking history"}], - "stop_reason": "end_turn", - "stop_sequence": None, - "usage": {"input_tokens": 30, "output_tokens": 6}, - } - ).encode() - ) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key=KEY) - response: Final = gateway.request( - "POST", - "/v1/messages", - { - "model": model, - "max_tokens": 64, - "thinking": {"type": "enabled", "budget_tokens": 1024}, - "tools": list(TOOLS), - "messages": list(history), - }, - ) - assert response.status_code == 200, response.text - body: Final = response.json() - assert body["id"] == identity, response.text - assert body["content"] == [{"type": "text", "text": "recovered without thinking history"}], response.text - assert body["stop_reason"] == "end_turn", response.text - assert [request.target for request in wire.drain()] == ["/v1/messages", "/v1/messages"] diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_tool_loop_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_tool_loop_wire.py deleted file mode 100644 index f3fb9786e02..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_tool_loop_wire.py +++ /dev/null @@ -1,165 +0,0 @@ -import uuid -from typing import Final - -from integration._support import claude_code as cc -from integration._support.client import Gateway, eventually -from integration._support.database import read_rows -from integration._support.wire import Reply, Request, wire_server -from pydantic import JsonValue - -_THINKING: Final = "need to read the file" -_SIGNATURE: Final = "sig_probe_1" -_USAGE: Final = {"input_tokens": 20, "output_tokens": 10} - - -def _diff(expected: dict[str, JsonValue], body: dict[str, JsonValue]) -> dict[str, JsonValue]: - return { - key: {"expected": expected.get(key), "upstream": body.get(key)} - for key in expected.keys() | body.keys() - if expected.get(key) != body.get(key) - } - - -def test_tool_loop_round_trips_thinking_tool_use_and_tool_result(gateway: Gateway) -> None: - identity1: Final = f"msg_tl1_{uuid.uuid4().hex}" - identity2: Final = f"msg_tl2_{uuid.uuid4().hex}" - 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", - ) - calls: Final = (("toolu_read_1", "Read", {"file_path": "/tmp/cc_probe/hello.txt"}),) - turn2: Final = cc.tool_loop_turn2( - turn1, - ( - {"type": "thinking", "thinking": _THINKING, "signature": _SIGNATURE}, - { - "type": "tool_use", - "id": "toolu_read_1", - "name": "Read", - "input": {"file_path": "/tmp/cc_probe/hello.txt"}, - }, - ), - (("toolu_read_1", "1\tPROBE\n2\t"),), - ) - first_expected: Final = {**turn1, "model": cc.FABLE} - second_expected: Final = {**turn2, "model": cc.FABLE} - - def respond(request: Request) -> Reply: - assert request.method == "POST" - assert request.target == "/v1/messages", request.target - body: Final = cc.JSON_OBJECT.validate_json(request.body) - if body == first_expected: - return Reply( - content_type="text/event-stream", - chunks=cc.tool_use_stream(identity1, cc.FABLE, _THINKING, _SIGNATURE, calls, _USAGE), - ) - assert body == second_expected, _diff(second_expected, body) - return Reply( - content_type="text/event-stream", - chunks=cc.text_stream(identity2, cc.FABLE, "PROBE", {"input_tokens": 30, "output_tokens": 3}), - ) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_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 - events: Final = cc.sse_events(response1.text) - assert [ - ( - event, - data.get("delta", {}).get( - "type", data.get("content_block", {}).get("type", data.get("delta", {}).get("stop_reason")) - ), - ) - for event, data in events - ] == [ - ("message_start", None), - ("content_block_start", "thinking"), - ("content_block_delta", "thinking_delta"), - ("content_block_delta", "signature_delta"), - ("content_block_stop", None), - ("content_block_start", "tool_use"), - ("content_block_delta", "input_json_delta"), - ("content_block_delta", "input_json_delta"), - ("content_block_stop", None), - ("message_delta", "tool_use"), - ("message_stop", None), - ] - assert events[5][1]["content_block"]["id"] == "toolu_read_1" - assert events[5][1]["content_block"]["name"] == "Read" - partial: Final = events[6][1]["delta"]["partial_json"] + events[7][1]["delta"]["partial_json"] - assert partial == '{"file_path": "/tmp/cc_probe/hello.txt"}' - response2: Final = gateway.request( - "POST", "/v1/messages", {**turn2, "model": model}, params={"beta": "true"}, headers=headers - ) - assert response2.status_code == 200, response2.text - events2: Final = cc.sse_events(response2.text) - assert events2[2][1]["delta"] == {"type": "text_delta", "text": "PROBE"} - assert events2[4][1]["delta"]["stop_reason"] == "end_turn" - bodies: Final = tuple(cc.JSON_OBJECT.validate_json(request.body) for request in wire.drain()) - assert bodies == (first_expected, second_expected), bodies - rows: Final = eventually( - lambda: read_rows( - 'SELECT prompt_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', - (identity2,), - ), - lambda values: len(values) == 1, - seconds=70, - ) - assert rows[0]["prompt_tokens"] == 30 - - -def test_parallel_tool_results_reach_anthropic_in_client_order(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 /tmp/cc_probe/world.txt and reply with both words", - ) - turn2: Final = cc.tool_loop_turn2( - turn1, - ( - {"type": "thinking", "thinking": _THINKING, "signature": _SIGNATURE}, - { - "type": "tool_use", - "id": "toolu_read_1", - "name": "Read", - "input": {"file_path": "/tmp/cc_probe/hello.txt"}, - }, - { - "type": "tool_use", - "id": "toolu_read_2", - "name": "Read", - "input": {"file_path": "/tmp/cc_probe/world.txt"}, - }, - ), - (("toolu_read_2", "1\tPROBE2\n2\t"), ("toolu_read_1", "1\tPROBE\n2\t")), - ) - - def respond(request: Request) -> Reply: - body: Final = cc.JSON_OBJECT.validate_json(request.body) - expected: Final = {**turn2, "model": cc.FABLE} - assert body == expected, _diff(expected, body) - results: Final = [block for block in body["messages"][3]["content"] if block["type"] == "tool_result"] - assert [block["tool_use_id"] for block in results] == ["toolu_read_2", "toolu_read_1"] - return Reply( - content_type="text/event-stream", - chunks=cc.text_stream(f"msg_mt_{uuid.uuid4().hex}", cc.FABLE, "PROBE PROBE2", _USAGE), - ) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_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 diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_web_search_citations_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_web_search_citations_wire.py deleted file mode 100644 index 66852e6e21b..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_web_search_citations_wire.py +++ /dev/null @@ -1,170 +0,0 @@ -import uuid -from typing import Final - -from integration._support import claude_code as cc -from integration._support.client import Gateway, eventually -from integration._support.database import read_rows -from integration._support.wire import Reply, Request, wire_server - -WEB_SEARCH_TOOL: Final = {"type": "web_search_20250305", "name": "web_search", "max_uses": 8} - - -def _web_search_stream(identity: str) -> tuple[bytes, ...]: - return ( - cc.sse_frame( - "message_start", - { - "type": "message_start", - "message": { - "id": identity, - "type": "message", - "role": "assistant", - "model": cc.FABLE, - "content": [], - "stop_reason": None, - "stop_sequence": None, - "usage": {"input_tokens": 20, "output_tokens": 1, "server_tool_use": {"web_search_requests": 1}}, - }, - }, - ), - cc.sse_frame( - "content_block_start", - { - "type": "content_block_start", - "index": 0, - "content_block": {"type": "server_tool_use", "id": "srvtoolu_1", "name": "web_search", "input": {}}, - }, - ), - cc.sse_frame( - "content_block_delta", - { - "type": "content_block_delta", - "index": 0, - "delta": {"type": "input_json_delta", "partial_json": '{"query": "current LiteLLM version"}'}, - }, - ), - cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}), - cc.sse_frame( - "content_block_start", - { - "type": "content_block_start", - "index": 1, - "content_block": { - "type": "web_search_tool_result", - "tool_use_id": "srvtoolu_1", - "content": [ - { - "type": "web_search_result", - "title": "litellm releases", - "url": "https://example.com/litellm", - "page_age": None, - "encrypted_content": "enc_ws_1", - } - ], - }, - }, - ), - cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 1}), - cc.sse_frame( - "content_block_start", - {"type": "content_block_start", "index": 2, "content_block": {"type": "text", "text": ""}}, - ), - cc.sse_frame( - "content_block_delta", - {"type": "content_block_delta", "index": 2, "delta": {"type": "text_delta", "text": "1.104.0"}}, - ), - cc.sse_frame( - "content_block_delta", - { - "type": "content_block_delta", - "index": 2, - "delta": { - "type": "citations_delta", - "citation": { - "type": "web_search_result_location", - "url": "https://example.com/litellm", - "title": "litellm releases", - "cited_text": "version 1.104.0", - "encrypted_index": "eidx_1", - }, - }, - }, - ), - cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 2}), - cc.sse_frame( - "message_delta", - { - "type": "message_delta", - "delta": {"stop_reason": "end_turn", "stop_sequence": None}, - "usage": {"output_tokens": 15, "server_tool_use": {"web_search_requests": 1}}, - }, - ), - cc.sse_frame("message_stop", {"type": "message_stop"}), - ) - - -def test_web_search_tool_passthrough_and_cited_response(gateway: Gateway) -> None: - identity: Final = f"msg_ws_{uuid.uuid4().hex}" - base: 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", - ) - request_body: Final = { - **base, - "tools": [*base["tools"], WEB_SEARCH_TOOL], - } - - def respond(request: Request) -> Reply: - assert request.method == "POST" - assert request.target == "/v1/messages", request.target - body: Final = cc.JSON_OBJECT.validate_json(request.body) - expected: Final = {**request_body, "model": cc.FABLE} - assert body == expected, { - key: (expected.get(key), body.get(key)) - for key in expected.keys() | body.keys() - if expected.get(key) != body.get(key) - } - assert body["tools"][-1] == WEB_SEARCH_TOOL - assert len({tool["name"] for tool in body["tools"]}) == len(body["tools"]), body["tools"] - return Reply(content_type="text/event-stream", chunks=_web_search_stream(identity)) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model( - model=f"anthropic/{cc.FABLE}", - api_base=wire.url, - api_key=cc.ANTHROPIC_API_KEY, - input_cost_per_token=1e-6, - output_cost_per_token=5e-6, - ) - 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 - events: Final = cc.sse_events(response.text) - started: Final = [ - (data["index"], data["content_block"]["type"]) for event, data in events if event == "content_block_start" - ] - assert started == [(0, "server_tool_use"), (1, "web_search_tool_result"), (2, "text")], started - citations: Final = [ - data["delta"] for event, data in events if data.get("delta", {}).get("type") == "citations_delta" - ] - assert len(citations) == 1 and citations[0]["citation"]["url"] == "https://example.com/litellm", citations - start_usage: Final = events[0][1]["message"]["usage"] - assert start_usage["server_tool_use"]["web_search_requests"] == 1, start_usage - assert len(wire.drain()) == 1 - rows: Final = eventually( - lambda: read_rows( - 'SELECT spend, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', - (identity,), - ), - lambda values: len(values) == 1, - seconds=70, - ) - token_cost: Final = 20 * 1e-6 + 15 * 5e-6 - assert float(rows[0]["spend"]) >= token_cost, dict(rows[0]) diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_wire.py deleted file mode 100644 index 7e9c5be227a..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_anthropic_wire.py +++ /dev/null @@ -1,206 +0,0 @@ -import json -import time -import uuid -from typing import Final - -import pytest -from integration._support.client import Gateway, eventually, object_value -from integration._support.database import read_rows -from integration._support.wire import Reply, Request, wire_server - - -@pytest.mark.covers( - "other.provider_wire.anthropic.tool_history_system_cache_and_internal_fields", - "quota_management.spend_tracking.cache_tokens.disjoint_classes_use_explicit_rates", -) -def test_anthropic_tool_history_and_cache_tokens_keep_wire_and_accounting_contracts(gateway: Gateway) -> None: - identity: Final = "anthropic-wire-" + uuid.uuid4().hex - tool_schema: Final = { - "type": "object", - "properties": {"x": {"type": "integer"}, "y": {"type": "integer"}}, - "required": ["x", "y"], - } - - def respond(request: Request) -> Reply: - assert request.method == "POST" and request.target == "/v1/messages" - assert request.headers["x-api-key"] == "synthetic-anthropic-key" - body: Final = json.loads(request.body) - assert body["model"] == "claude-sonnet-4-5-20250929" - assert body["system"] == [{"type": "text", "text": "synthetic policy", "cache_control": {"type": "ephemeral"}}] - assert body["tools"][0]["name"] == "add" and body["tools"][0]["input_schema"] == tool_schema - assert body["max_tokens"] == 16 - assert not {"timeout", "stream_chunk_size", "litellm_params", "litellm_metadata", "rpm", "tpm"}.intersection( - body - ) - messages: Final = body["messages"] - assert [message["role"] for message in messages] == ["user", "assistant", "user"] - assert messages[0]["content"] == [{"type": "text", "text": "first"}] - assert messages[1]["content"] == [ - {"type": "tool_use", "id": "history-call", "name": "add", "input": {"x": 1, "y": 2}} - ] - assert messages[2]["content"] == [ - {"type": "tool_result", "tool_use_id": "history-call", "content": "3"}, - {"type": "text", "text": "next"}, - ] - return Reply( - body=json.dumps( - { - "id": identity, - "type": "message", - "role": "assistant", - "model": "claude-sonnet-4-5-20250929", - "content": [{"type": "tool_use", "id": "next-call", "name": "add", "input": {"x": 3, "y": 4}}], - "stop_reason": "tool_use", - "stop_sequence": None, - "usage": { - "input_tokens": 10, - "output_tokens": 4, - "cache_read_input_tokens": 5, - "cache_creation_input_tokens": 7, - }, - } - ).encode() - ) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model( - model="anthropic/claude-sonnet-4-5-20250929", - api_base=wire.url, - api_key="synthetic-anthropic-key", - input_cost_per_token=0.001, - output_cost_per_token=0.002, - cache_read_input_token_cost=0.0001, - cache_creation_input_token_cost=0.002, - ) - response: Final = gateway.request( - "POST", - "/v1/chat/completions", - { - "model": model, - "max_tokens": 16, - "timeout": 5, - "messages": [ - { - "role": "system", - "content": [ - {"type": "text", "text": "synthetic policy", "cache_control": {"type": "ephemeral"}} - ], - }, - {"role": "user", "content": "first"}, - { - "role": "assistant", - "tool_calls": [ - { - "id": "history-call", - "type": "function", - "function": {"name": "add", "arguments": '{"x":1,"y":2}'}, - } - ], - }, - {"role": "tool", "tool_call_id": "history-call", "content": "3"}, - {"role": "user", "content": "next"}, - ], - "tools": [{"type": "function", "function": {"name": "add", "parameters": tool_schema}}], - }, - ) - assert response.status_code == 200, response.text - body: Final = response.json() - assert body["id"].startswith("chatcmpl-") - assert body["choices"][0]["finish_reason"] == "tool_calls" - tool: Final = body["choices"][0]["message"]["tool_calls"][0] - assert tool["id"] == "next-call" and tool["function"]["name"] == "add" - assert json.loads(tool["function"]["arguments"]) == {"x": 3, "y": 4} - assert body["usage"]["prompt_tokens"] == 22 and body["usage"]["completion_tokens"] == 4 - assert len(wire.drain()) == 1 - rows: Final = eventually( - lambda: read_rows( - 'SELECT spend, prompt_tokens, completion_tokens, metadata FROM "LiteLLM_SpendLogs" WHERE request_id=%s', - (body["id"],), - ), - lambda values: len(values) == 1, - seconds=70, - ) - assert float(rows[0]["spend"]) == pytest.approx(10 * 0.001 + 5 * 0.0001 + 7 * 0.002 + 4 * 0.002) - assert rows[0]["prompt_tokens"] == 22 and rows[0]["completion_tokens"] == 4 - metadata: Final = rows[0]["metadata"] - parsed: Final = json.loads(metadata) if isinstance(metadata, str) else object_value(metadata) - assert parsed["cost_breakdown"]["input_cost"] == pytest.approx(0.0245) - assert parsed["cost_breakdown"]["output_cost"] == pytest.approx(0.008) - - -@pytest.mark.covers("other.provider_wire.anthropic.bare_string_content_item_is_client_error") -@pytest.mark.parametrize( - "text", [pytest.param("what type of file is this?", id="type_word"), pytest.param("hello", id="plain")] -) -def test_anthropic_bare_string_content_item_is_rejected_as_client_error_before_the_wire( - gateway: Gateway, text: str -) -> None: - def respond(request: Request) -> Reply: - raise AssertionError(f"upstream must not be reached: {request.target}") - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model( - model="anthropic/claude-sonnet-4-5-20250929", api_base=wire.url, api_key="synthetic-anthropic-key" - ) - response: Final = gateway.request( - "POST", - "/v1/chat/completions", - {"model": model, "max_tokens": 16, "timeout": 5, "messages": [{"role": "system", "content": [text]}]}, - ) - assert response.status_code == 400, response.text - assert wire.drain() == () - - -@pytest.mark.covers("other.provider_wire.anthropic.messages_request_timeout_reaches_transport") -def test_anthropic_messages_slow_upstream_is_cut_off_at_the_deployment_request_timeout(gateway: Gateway) -> None: - identity: Final = "anthropic-timeout-" + uuid.uuid4().hex - prompt: Final = f"slow answer {identity}" - - def respond(request: Request) -> Reply: - assert request.method == "POST" and request.target == "/v1/messages" - assert request.headers["x-api-key"] == "synthetic-anthropic-key" - body: Final = json.loads(request.body) - assert body["model"] == "claude-sonnet-4-5-20250929" - assert body["max_tokens"] == 16 - assert body["messages"] == [{"role": "user", "content": prompt}] - assert not { - "timeout", - "request_timeout", - "stream_chunk_size", - "litellm_params", - "litellm_metadata", - "rpm", - "tpm", - }.intersection(body) - time.sleep(1.5) - return Reply( - body=json.dumps( - { - "id": identity, - "type": "message", - "role": "assistant", - "model": "claude-sonnet-4-5-20250929", - "content": [{"type": "text", "text": "late"}], - "stop_reason": "end_turn", - "stop_sequence": None, - "usage": {"input_tokens": 3, "output_tokens": 1}, - } - ).encode() - ) - - with wire_server(respond) as wire, gateway.scenario() as scenario: - model: Final = scenario.model( - model="anthropic/claude-sonnet-4-5-20250929", - api_base=wire.url, - api_key="synthetic-anthropic-key", - request_timeout=0.3, - ) - response: Final = gateway.request( - "POST", - "/v1/messages", - {"model": model, "max_tokens": 16, "messages": [{"role": "user", "content": prompt}]}, - headers={"anthropic-version": "2023-06-01"}, - ) - assert response.status_code == 408, response.text - assert "Timeout" in response.json()["error"]["message"], response.text - assert eventually(wire.drain, lambda requests: len(requests) == 1, seconds=5, return_last_on_timeout=True) diff --git a/tests/integration/messages_endpoint/providers/anthropic/test_websearch_interception_wire.py b/tests/integration/messages_endpoint/providers/anthropic/test_websearch_interception_wire.py deleted file mode 100644 index a6f098cf64f..00000000000 --- a/tests/integration/messages_endpoint/providers/anthropic/test_websearch_interception_wire.py +++ /dev/null @@ -1,409 +0,0 @@ -import json -from pathlib import Path -from typing import Final - -import pytest -import yaml -from integration._support.client import Gateway -from integration._support.process import owned_proxy -from integration._support.wire import Reply, Request, wire_server - -BEDROCK_MODEL: Final = "us.anthropic.claude-haiku-4-5-20251001-v1:0" -INVOKE_TARGET: Final = f"/model/{BEDROCK_MODEL}/invoke" -SEARCH_TARGET: Final = "/tavily/search" -SEARCH_RESULT: Final = { - "title": "Synthetic result", - "url": "https://example.test/result", - "content": "the snippet text", -} - - -def sse_events(text: str) -> tuple[tuple[str, dict[str, object]], ...]: - frames: Final = tuple(frame for frame in text.split("\n\n") if frame.strip()) - return tuple( - ( - next(line.removeprefix("event: ") for line in frame.splitlines() if line.startswith("event: ")), - json.loads(next(line.removeprefix("data: ") for line in frame.splitlines() if line.startswith("data: "))), - ) - for frame in frames - ) - - -@pytest.mark.covers("other.provider_wire.bedrock.websearch_interception_streamed_capped_turn_ends_with_native_results") -def test_streamed_web_search_turn_capped_by_max_agentic_loops_ends_turn_with_snippets_and_ordered_blocks( - gateway: Gateway, tmp_path: Path -) -> None: - def respond(request: Request) -> Reply: - assert request.method == "POST", request.target - body: Final = json.loads(request.body) - if request.target == SEARCH_TARGET: - assert request.headers["authorization"] == "Bearer synthetic-tavily-key" - assert body["query"] == "query-0", body - return Reply(body=json.dumps({"query": "query-0", "results": [SEARCH_RESULT]}).encode()) - assert request.target == INVOKE_TARGET - assert request.headers["authorization"] == "Bearer synthetic-bedrock-token" - assert [tool["name"] for tool in body["tools"]] == ["litellm_web_search"], body["tools"] - assert "stream" not in body, body - depth: Final = sum( - 1 - for message in body["messages"] - if isinstance(message["content"], list) - for block in message["content"] - if block["type"] == "tool_result" - ) - if depth == 1: - assert body["messages"][2]["content"] == [ - { - "type": "tool_result", - "tool_use_id": "toolu_0", - "content": "Title: Synthetic result\nURL: https://example.test/result\nSnippet: the snippet text", - } - ], body["messages"] - return Reply( - body=json.dumps( - { - "id": f"msg_{depth}", - "type": "message", - "role": "assistant", - "model": BEDROCK_MODEL, - "content": [ - {"type": "text", "text": f"turn-{depth}"}, - { - "type": "tool_use", - "id": f"toolu_{depth}", - "name": "litellm_web_search", - "input": {"query": f"query-{depth}"}, - }, - ], - "stop_reason": "tool_use", - "stop_sequence": None, - "usage": {"input_tokens": 10, "output_tokens": 4}, - } - ).encode() - ) - - with wire_server(respond) as wire: - config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) - config["search_tools"] = [ - { - "search_tool_name": "integration-search", - "litellm_params": { - "search_provider": "tavily", - "api_key": "synthetic-tavily-key", - "api_base": wire.url + "/tavily", - }, - } - ] - config["litellm_settings"].update( - { - "callbacks": ["websearch_interception"], - "websearch_interception_params": { - "enabled_providers": ["bedrock"], - "search_tool_name": "integration-search", - "max_agentic_loops": 1, - }, - } - ) - path: Final = tmp_path / "websearch.yaml" - path.write_text(yaml.safe_dump(config)) - with owned_proxy(gateway, tmp_path, {}, config=path) as candidate, candidate.scenario() as scenario: - model: Final = scenario.model( - model=f"bedrock/{BEDROCK_MODEL}", - api_key="synthetic-bedrock-token", - api_base=wire.url, - aws_region_name="us-east-1", - aws_bedrock_runtime_endpoint=wire.url, - ) - response: Final = candidate.request( - "POST", - "/v1/messages", - { - "model": model, - "max_tokens": 64, - "stream": True, - "messages": [{"role": "user", "content": "search control"}], - "tools": [{"type": "web_search_20250305", "name": "web_search"}], - }, - ) - assert response.status_code == 200, response.text - events: Final = sse_events(response.text) - assert [name for name, _ in events][:1] == ["message_start"], response.text - assert [name for name, _ in events][-2:] == ["message_delta", "message_stop"], response.text - for position, (name, event) in enumerate(events): - if name == "content_block_stop": - assert event["index"] in { - earlier_event["index"] - for earlier, earlier_event in events[:position] - if earlier == "content_block_start" - }, response.text - started: Final = tuple(event["content_block"] for name, event in events if name == "content_block_start") - search_ids: Final = tuple(block["id"] for block in started if block["type"] == "server_tool_use") - assert search_ids and all(search_id.startswith("srvtoolu_") for search_id in search_ids), response.text - assert started[-1] == {"type": "text", "text": ""}, response.text - assert started[:-1] == tuple( - block - for search_id in search_ids - for block in ( - {"type": "server_tool_use", "id": search_id, "name": "web_search", "input": {"query": "query-0"}}, - { - "type": "web_search_tool_result", - "tool_use_id": search_id, - "content": [ - { - "type": "web_search_result", - "url": "https://example.test/result", - "title": "Synthetic result", - "page_age": None, - "encrypted_content": "", - "snippet": "the snippet text", - } - ], - }, - ) - ), response.text - assert ( - "".join(event["delta"]["text"] for name, event in events if name == "content_block_delta") == "turn-1" - ), response.text - assert [event["delta"]["stop_reason"] for name, event in events if name == "message_delta"] == [ - "end_turn" - ], response.text - assert "litellm_web_search" not in response.text, response.text - assert [request.target for request in wire.drain()] == [INVOKE_TARGET, SEARCH_TARGET, INVOKE_TARGET] - - -import threading -import uuid -from typing import Final -from urllib.parse import parse_qs, urlsplit - -import httpx -import pytest -from integration._support.client import Gateway, eventually - -_QUERY: Final = "integration capped search" -_TEXT_BLOCK: Final = {"type": "text", "text": "searching once more"} -_NOT_INTERCEPTED: Final = "native tool reached the provider" -_FINAL_BLOCK: Final = {"type": "text", "text": "answered from the stored backend"} -_OWNED_RESULT_TEXT: Final = "Title: Owned result\nURL: https://owned.invalid/a\nSnippet: owned snippet" -_SEARCH_RESULT_BLOCK: Final = { - "type": "web_search_result", - "url": "https://owned.invalid/a", - "title": "Owned result", - "page_age": None, - "encrypted_content": "", - "snippet": "owned snippet", -} - - -def _search_tool_use(identity: str) -> dict[str, object]: - return {"type": "tool_use", "id": identity, "name": "litellm_web_search", "input": {"query": _QUERY}} - - -def _anthropic_reply(identity: str, content: list[dict[str, object]], stop_reason: str) -> Reply: - return Reply( - body=json.dumps( - { - "id": identity, - "type": "message", - "role": "assistant", - "model": "claude-sonnet-4-5-20250929", - "content": content, - "stop_reason": stop_reason, - "stop_sequence": None, - "usage": {"input_tokens": 10, "output_tokens": 4}, - } - ).encode() - ) - - -@pytest.mark.covers( - "other.provider_wire.anthropic.websearch_interception_capped_loop_ends_turn_without_internal_tool_use" -) -def test_capped_websearch_interception_loop_ends_turn_instead_of_exposing_internal_tool_use( - gateway: Gateway, tmp_path: Path -) -> None: - identity: Final = "websearch-wire-" + uuid.uuid4().hex - searched: Final = threading.Event() - - def respond(request: Request) -> Reply: - parts: Final = urlsplit(request.target) - if request.method == "GET" and parts.path == "/search": - assert parse_qs(parts.query)["q"] == [_QUERY], request.target - searched.set() - return Reply( - body=json.dumps( - { - "results": [ - {"title": "Owned result", "url": "https://owned.invalid/a", "content": "owned snippet"} - ] - } - ).encode() - ) - assert request.method == "POST" and parts.path == "/v1/messages", request.target - body: Final = json.loads(request.body) - if any(tool.get("type") == "web_search_20250305" for tool in body["tools"]): - return _anthropic_reply(identity, [{"type": "text", "text": _NOT_INTERCEPTED}], "end_turn") - assert [tool["name"] for tool in body["tools"]] == ["litellm_web_search"], body["tools"] - return _anthropic_reply(identity, [_TEXT_BLOCK, _search_tool_use(identity)], "tool_use") - - def send(candidate: Gateway, model: str) -> httpx.Response: - return candidate.request( - "POST", - "/v1/messages", - { - "model": model, - "max_tokens": 64, - "messages": [{"role": "user", "content": identity + " attempt " + uuid.uuid4().hex}], - "tools": [{"type": "web_search_20250305", "name": "web_search", "max_uses": 3}], - }, - ) - - def searched_through_proxy(response: httpx.Response) -> bool: - return searched.is_set() and _NOT_INTERCEPTED not in response.text - - with wire_server(respond) as wire: - config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) - config["search_tools"] = [ - { - "search_tool_name": "integration-searxng", - "litellm_params": {"search_provider": "searxng", "api_base": wire.url}, - } - ] - config["litellm_settings"].update( - { - "callbacks": ["websearch_interception"], - "websearch_interception_params": { - "enabled": True, - "enabled_providers": ["anthropic"], - "search_tool_name": "integration-searxng", - }, - } - ) - path: Final = tmp_path / "websearch.yaml" - path.write_text(yaml.safe_dump(config)) - with owned_proxy(gateway, tmp_path, {}, config=path) as candidate, candidate.scenario() as scenario: - model: Final = scenario.model( - model="anthropic/claude-sonnet-4-5-20250929", api_base=wire.url, api_key="synthetic-anthropic-key" - ) - response: Final = eventually(lambda: send(candidate, model), searched_through_proxy, seconds=40) - assert response.status_code == 200, response.text - body: Final = response.json() - assert body["stop_reason"] == "end_turn", response.text - content: Final = body["content"] - assert [block["type"] for block in content] == ["server_tool_use", "web_search_tool_result", "text"], ( - response.text - ) - assert content[0]["name"] == "web_search" and content[0]["input"] == {"query": _QUERY}, response.text - assert content[1]["tool_use_id"] == content[0]["id"], response.text - assert content[1]["content"] == [_SEARCH_RESULT_BLOCK], response.text - assert content[2] == _TEXT_BLOCK, response.text - targets: Final = tuple((request.method, urlsplit(request.target).path) for request in wire.drain()) - assert targets[-3:] == (("POST", "/v1/messages"), ("GET", "/search"), ("POST", "/v1/messages")), targets - - -@pytest.mark.covers("other.provider_wire.anthropic.websearch_interception_uses_database_search_tool_backend") -def test_database_created_search_tool_backend_receives_the_intercepted_query_over_a_same_named_config_tool( - gateway: Gateway, tmp_path: Path -) -> None: - identity: Final = "websearch-db-" + uuid.uuid4().hex - tool_name: Final = "integration-db-searxng-" + uuid.uuid4().hex - searched: Final = threading.Event() - - def respond(request: Request) -> Reply: - parts: Final = urlsplit(request.target) - if request.method == "GET" and parts.path == "/database/search": - assert parse_qs(parts.query)["q"] == [_QUERY], request.target - searched.set() - return Reply( - body=json.dumps( - { - "results": [ - {"title": "Owned result", "url": "https://owned.invalid/a", "content": "owned snippet"} - ] - } - ).encode() - ) - assert request.method == "POST" and parts.path == "/v1/messages", request.target - body: Final = json.loads(request.body) - if any(tool.get("type") == "web_search_20250305" for tool in body["tools"]): - return _anthropic_reply(identity, [{"type": "text", "text": _NOT_INTERCEPTED}], "end_turn") - assert [tool["name"] for tool in body["tools"]] == ["litellm_web_search"], body["tools"] - results: Final = [ - block - for message in body["messages"] - if isinstance(message["content"], list) - for block in message["content"] - if block["type"] == "tool_result" - ] - if not results: - return _anthropic_reply(identity, [_TEXT_BLOCK, _search_tool_use(identity)], "tool_use") - assert results == [{"type": "tool_result", "tool_use_id": identity, "content": _OWNED_RESULT_TEXT}], results - return _anthropic_reply(identity, [_FINAL_BLOCK], "end_turn") - - def send(candidate: Gateway, model: str) -> httpx.Response: - return candidate.request( - "POST", - "/v1/messages", - { - "model": model, - "max_tokens": 64, - "messages": [{"role": "user", "content": identity + " attempt " + uuid.uuid4().hex}], - "tools": [{"type": "web_search_20250305", "name": "web_search", "max_uses": 3}], - }, - ) - - def searched_through_proxy(response: httpx.Response) -> bool: - return searched.is_set() and _NOT_INTERCEPTED not in response.text - - with wire_server(respond) as wire, gateway.scenario() as scenario: - created: Final = gateway.post( - "/search_tools", - { - "search_tool": { - "search_tool_name": tool_name, - "litellm_params": {"search_provider": "searxng", "api_base": wire.url + "/database"}, - } - }, - ) - scenario.cleanups.callback(gateway.request, "DELETE", f"/search_tools/{created['search_tool_id']}") - config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) - config["search_tools"] = [ - { - "search_tool_name": tool_name, - "litellm_params": {"search_provider": "searxng", "api_base": wire.url + "/config"}, - } - ] - config["litellm_settings"].update( - { - "callbacks": ["websearch_interception"], - "websearch_interception_params": { - "enabled": True, - "enabled_providers": ["anthropic"], - "search_tool_name": tool_name, - }, - } - ) - path: Final = tmp_path / "websearch-db.yaml" - path.write_text(yaml.safe_dump(config)) - environment: Final = {"ANTHROPIC_API_BASE": wire.url} - with owned_proxy(gateway, tmp_path, environment, config=path) as candidate, candidate.scenario() as models: - model: Final = models.model( - model="anthropic/claude-sonnet-4-5-20250929", api_base=wire.url, api_key="synthetic-anthropic-key" - ) - response: Final = eventually(lambda: send(candidate, model), searched_through_proxy, seconds=40) - assert response.status_code == 200, response.text - body: Final = response.json() - assert body["stop_reason"] == "end_turn", response.text - assert body["content"][-1] == _FINAL_BLOCK, response.text - found: Final = [ - (result["url"], result["title"]) - for block in body["content"] - if block["type"] == "web_search_tool_result" - for result in block["content"] - ] - assert found == [("https://owned.invalid/a", "Owned result")], response.text - assert "litellm_web_search" not in response.text, response.text - targets: Final = tuple((request.method, urlsplit(request.target).path) for request in wire.drain()) - assert targets[-3:] == (("POST", "/v1/messages"), ("GET", "/database/search"), ("POST", "/v1/messages")), ( - targets - )