|
|
@@ -0,0 +1,807 @@
|
|
|
+#!/usr/bin/env python3
|
|
|
+import pytest
|
|
|
+import base64
|
|
|
+import requests
|
|
|
+
|
|
|
+from utils import *
|
|
|
+
|
|
|
+server: ServerProcess
|
|
|
+
|
|
|
+
|
|
|
+def get_test_image_base64() -> str:
|
|
|
+ """Get a test image in base64 format"""
|
|
|
+ # Use the same test image as test_vision_api.py
|
|
|
+ IMG_URL = "https://huggingface.co/ggml-org/tinygemma3-GGUF/resolve/main/test/11_truck.png"
|
|
|
+ response = requests.get(IMG_URL)
|
|
|
+ response.raise_for_status()
|
|
|
+ return base64.b64encode(response.content).decode("utf-8")
|
|
|
+
|
|
|
+@pytest.fixture(autouse=True)
|
|
|
+def create_server():
|
|
|
+ global server
|
|
|
+ server = ServerPreset.tinyllama2()
|
|
|
+ server.model_alias = "tinyllama-2-anthropic"
|
|
|
+ server.server_port = 8082
|
|
|
+ server.n_slots = 1
|
|
|
+ server.n_ctx = 8192
|
|
|
+ server.n_batch = 2048
|
|
|
+
|
|
|
+
|
|
|
+@pytest.fixture
|
|
|
+def vision_server():
|
|
|
+ """Separate fixture for vision tests that require multimodal support"""
|
|
|
+ global server
|
|
|
+ server = ServerPreset.tinygemma3()
|
|
|
+ server.offline = False # Allow downloading the model
|
|
|
+ server.model_alias = "tinygemma3-anthropic"
|
|
|
+ server.server_port = 8083 # Different port to avoid conflicts
|
|
|
+ server.n_slots = 1
|
|
|
+ return server
|
|
|
+
|
|
|
+
|
|
|
+# Basic message tests
|
|
|
+
|
|
|
+def test_anthropic_messages_basic():
|
|
|
+ """Test basic Anthropic messages endpoint"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 50,
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Say hello"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200, f"Expected 200, got {res.status_code}"
|
|
|
+ assert res.body["type"] == "message", f"Expected type 'message', got {res.body.get('type')}"
|
|
|
+ assert res.body["role"] == "assistant", f"Expected role 'assistant', got {res.body.get('role')}"
|
|
|
+ assert "content" in res.body, "Missing 'content' field"
|
|
|
+ assert isinstance(res.body["content"], list), "Content should be an array"
|
|
|
+ assert len(res.body["content"]) > 0, "Content array should not be empty"
|
|
|
+ assert res.body["content"][0]["type"] == "text", "First content block should be text"
|
|
|
+ assert "text" in res.body["content"][0], "Text content block missing 'text' field"
|
|
|
+ assert res.body["stop_reason"] in ["end_turn", "max_tokens"], f"Invalid stop_reason: {res.body.get('stop_reason')}"
|
|
|
+ assert "usage" in res.body, "Missing 'usage' field"
|
|
|
+ assert "input_tokens" in res.body["usage"], "Missing usage.input_tokens"
|
|
|
+ assert "output_tokens" in res.body["usage"], "Missing usage.output_tokens"
|
|
|
+ assert isinstance(res.body["usage"]["input_tokens"], int), "input_tokens should be integer"
|
|
|
+ assert isinstance(res.body["usage"]["output_tokens"], int), "output_tokens should be integer"
|
|
|
+ assert res.body["usage"]["output_tokens"] > 0, "Should have generated some tokens"
|
|
|
+ # Anthropic API should NOT include timings
|
|
|
+ assert "timings" not in res.body, "Anthropic API should not include timings field"
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_messages_with_system():
|
|
|
+ """Test messages with system prompt"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 50,
|
|
|
+ "system": "You are a helpful assistant.",
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Hello"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+ assert len(res.body["content"]) > 0
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_messages_multipart_content():
|
|
|
+ """Test messages with multipart content blocks"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 50,
|
|
|
+ "messages": [
|
|
|
+ {
|
|
|
+ "role": "user",
|
|
|
+ "content": [
|
|
|
+ {"type": "text", "text": "What is"},
|
|
|
+ {"type": "text", "text": " the answer?"}
|
|
|
+ ]
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_messages_conversation():
|
|
|
+ """Test multi-turn conversation"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 50,
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Hello"},
|
|
|
+ {"role": "assistant", "content": "Hi there!"},
|
|
|
+ {"role": "user", "content": "How are you?"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+
|
|
|
+
|
|
|
+# Streaming tests
|
|
|
+
|
|
|
+def test_anthropic_messages_streaming():
|
|
|
+ """Test streaming messages"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_stream_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 30,
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Say hello"}
|
|
|
+ ],
|
|
|
+ "stream": True
|
|
|
+ })
|
|
|
+
|
|
|
+ events = []
|
|
|
+ for data in res:
|
|
|
+ # Each event should have type and other fields
|
|
|
+ assert "type" in data, f"Missing 'type' in event: {data}"
|
|
|
+ events.append(data)
|
|
|
+
|
|
|
+ # Verify event sequence
|
|
|
+ event_types = [e["type"] for e in events]
|
|
|
+ assert "message_start" in event_types, "Missing message_start event"
|
|
|
+ assert "content_block_start" in event_types, "Missing content_block_start event"
|
|
|
+ assert "content_block_delta" in event_types, "Missing content_block_delta event"
|
|
|
+ assert "content_block_stop" in event_types, "Missing content_block_stop event"
|
|
|
+ assert "message_delta" in event_types, "Missing message_delta event"
|
|
|
+ assert "message_stop" in event_types, "Missing message_stop event"
|
|
|
+
|
|
|
+ # Check message_start structure
|
|
|
+ message_start = next(e for e in events if e["type"] == "message_start")
|
|
|
+ assert "message" in message_start, "message_start missing 'message' field"
|
|
|
+ assert message_start["message"]["type"] == "message"
|
|
|
+ assert message_start["message"]["role"] == "assistant"
|
|
|
+ assert message_start["message"]["content"] == []
|
|
|
+ assert "usage" in message_start["message"]
|
|
|
+ assert message_start["message"]["usage"]["input_tokens"] > 0
|
|
|
+
|
|
|
+ # Check content_block_start
|
|
|
+ block_start = next(e for e in events if e["type"] == "content_block_start")
|
|
|
+ assert "index" in block_start, "content_block_start missing 'index'"
|
|
|
+ assert block_start["index"] == 0, "First content block should be at index 0"
|
|
|
+ assert "content_block" in block_start
|
|
|
+ assert block_start["content_block"]["type"] == "text"
|
|
|
+
|
|
|
+ # Check content_block_delta
|
|
|
+ deltas = [e for e in events if e["type"] == "content_block_delta"]
|
|
|
+ assert len(deltas) > 0, "Should have at least one content_block_delta"
|
|
|
+ for delta in deltas:
|
|
|
+ assert "index" in delta
|
|
|
+ assert "delta" in delta
|
|
|
+ assert delta["delta"]["type"] == "text_delta"
|
|
|
+ assert "text" in delta["delta"]
|
|
|
+
|
|
|
+ # Check content_block_stop
|
|
|
+ block_stop = next(e for e in events if e["type"] == "content_block_stop")
|
|
|
+ assert "index" in block_stop
|
|
|
+ assert block_stop["index"] == 0
|
|
|
+
|
|
|
+ # Check message_delta
|
|
|
+ message_delta = next(e for e in events if e["type"] == "message_delta")
|
|
|
+ assert "delta" in message_delta
|
|
|
+ assert "stop_reason" in message_delta["delta"]
|
|
|
+ assert message_delta["delta"]["stop_reason"] in ["end_turn", "max_tokens"]
|
|
|
+ assert "usage" in message_delta
|
|
|
+ assert message_delta["usage"]["output_tokens"] > 0
|
|
|
+
|
|
|
+ # Check message_stop
|
|
|
+ message_stop = next(e for e in events if e["type"] == "message_stop")
|
|
|
+ # message_stop should NOT have timings for Anthropic API
|
|
|
+ assert "timings" not in message_stop, "Anthropic streaming should not include timings"
|
|
|
+
|
|
|
+
|
|
|
+# Token counting tests
|
|
|
+
|
|
|
+def test_anthropic_count_tokens():
|
|
|
+ """Test token counting endpoint"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages/count_tokens", data={
|
|
|
+ "model": "test",
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Hello world"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert "input_tokens" in res.body
|
|
|
+ assert isinstance(res.body["input_tokens"], int)
|
|
|
+ assert res.body["input_tokens"] > 0
|
|
|
+ # Should only have input_tokens, no other fields
|
|
|
+ assert "output_tokens" not in res.body
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_count_tokens_with_system():
|
|
|
+ """Test token counting with system prompt"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages/count_tokens", data={
|
|
|
+ "model": "test",
|
|
|
+ "system": "You are a helpful assistant.",
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Hello"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["input_tokens"] > 0
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_count_tokens_no_max_tokens():
|
|
|
+ """Test that count_tokens doesn't require max_tokens"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ # max_tokens is NOT required for count_tokens
|
|
|
+ res = server.make_request("POST", "/v1/messages/count_tokens", data={
|
|
|
+ "model": "test",
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Hello"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert "input_tokens" in res.body
|
|
|
+
|
|
|
+
|
|
|
+# Tool use tests
|
|
|
+
|
|
|
+def test_anthropic_tool_use_basic():
|
|
|
+ """Test basic tool use"""
|
|
|
+ server.jinja = True
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 200,
|
|
|
+ "tools": [{
|
|
|
+ "name": "get_weather",
|
|
|
+ "description": "Get the current weather in a location",
|
|
|
+ "input_schema": {
|
|
|
+ "type": "object",
|
|
|
+ "properties": {
|
|
|
+ "location": {
|
|
|
+ "type": "string",
|
|
|
+ "description": "City name"
|
|
|
+ }
|
|
|
+ },
|
|
|
+ "required": ["location"]
|
|
|
+ }
|
|
|
+ }],
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "What's the weather in Paris?"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+ assert len(res.body["content"]) > 0
|
|
|
+
|
|
|
+ # Check if model used the tool (it might not always, depending on the model)
|
|
|
+ content_types = [block.get("type") for block in res.body["content"]]
|
|
|
+
|
|
|
+ if "tool_use" in content_types:
|
|
|
+ # Model used the tool
|
|
|
+ assert res.body["stop_reason"] == "tool_use"
|
|
|
+
|
|
|
+ # Find the tool_use block
|
|
|
+ tool_block = next(b for b in res.body["content"] if b.get("type") == "tool_use")
|
|
|
+ assert "id" in tool_block
|
|
|
+ assert "name" in tool_block
|
|
|
+ assert tool_block["name"] == "get_weather"
|
|
|
+ assert "input" in tool_block
|
|
|
+ assert isinstance(tool_block["input"], dict)
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_tool_result():
|
|
|
+ """Test sending tool results back
|
|
|
+
|
|
|
+ This test verifies that tool_result blocks are properly converted to
|
|
|
+ role="tool" messages internally. Without proper conversion, this would
|
|
|
+ fail with a 500 error: "unsupported content[].type" because tool_result
|
|
|
+ blocks would remain in the user message content array.
|
|
|
+ """
|
|
|
+ server.jinja = True
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 100,
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "What's the weather?"},
|
|
|
+ {
|
|
|
+ "role": "assistant",
|
|
|
+ "content": [
|
|
|
+ {
|
|
|
+ "type": "tool_use",
|
|
|
+ "id": "test123",
|
|
|
+ "name": "get_weather",
|
|
|
+ "input": {"location": "Paris"}
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "role": "user",
|
|
|
+ "content": [
|
|
|
+ {
|
|
|
+ "type": "tool_result",
|
|
|
+ "tool_use_id": "test123",
|
|
|
+ "content": "The weather is sunny, 25°C"
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ # This would be 500 with the old bug where tool_result blocks weren't converted
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+ # Model should respond to the tool result
|
|
|
+ assert len(res.body["content"]) > 0
|
|
|
+ assert res.body["content"][0]["type"] == "text"
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_tool_result_with_text():
|
|
|
+ """Test tool result mixed with text content
|
|
|
+
|
|
|
+ This tests the edge case where a user message contains both text and
|
|
|
+ tool_result blocks. The server must properly split these into separate
|
|
|
+ messages: a user message with text, followed by tool messages.
|
|
|
+ Without proper handling, this would fail with 500: "unsupported content[].type"
|
|
|
+ """
|
|
|
+ server.jinja = True
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 100,
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "What's the weather?"},
|
|
|
+ {
|
|
|
+ "role": "assistant",
|
|
|
+ "content": [
|
|
|
+ {
|
|
|
+ "type": "tool_use",
|
|
|
+ "id": "tool_1",
|
|
|
+ "name": "get_weather",
|
|
|
+ "input": {"location": "Paris"}
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "role": "user",
|
|
|
+ "content": [
|
|
|
+ {"type": "text", "text": "Here are the results:"},
|
|
|
+ {
|
|
|
+ "type": "tool_result",
|
|
|
+ "tool_use_id": "tool_1",
|
|
|
+ "content": "Sunny, 25°C"
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+ assert len(res.body["content"]) > 0
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_tool_result_error():
|
|
|
+ """Test tool result with error flag"""
|
|
|
+ server.jinja = True
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 100,
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Get the weather"},
|
|
|
+ {
|
|
|
+ "role": "assistant",
|
|
|
+ "content": [
|
|
|
+ {
|
|
|
+ "type": "tool_use",
|
|
|
+ "id": "test123",
|
|
|
+ "name": "get_weather",
|
|
|
+ "input": {"location": "InvalidCity"}
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "role": "user",
|
|
|
+ "content": [
|
|
|
+ {
|
|
|
+ "type": "tool_result",
|
|
|
+ "tool_use_id": "test123",
|
|
|
+ "is_error": True,
|
|
|
+ "content": "City not found"
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_tool_streaming():
|
|
|
+ """Test streaming with tool use"""
|
|
|
+ server.jinja = True
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_stream_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 200,
|
|
|
+ "stream": True,
|
|
|
+ "tools": [{
|
|
|
+ "name": "calculator",
|
|
|
+ "description": "Calculate math",
|
|
|
+ "input_schema": {
|
|
|
+ "type": "object",
|
|
|
+ "properties": {
|
|
|
+ "expression": {"type": "string"}
|
|
|
+ },
|
|
|
+ "required": ["expression"]
|
|
|
+ }
|
|
|
+ }],
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Calculate 2+2"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ events = []
|
|
|
+ for data in res:
|
|
|
+ events.append(data)
|
|
|
+
|
|
|
+ event_types = [e["type"] for e in events]
|
|
|
+
|
|
|
+ # Should have basic events
|
|
|
+ assert "message_start" in event_types
|
|
|
+ assert "message_stop" in event_types
|
|
|
+
|
|
|
+ # If tool was used, check for proper tool streaming
|
|
|
+ if any(e.get("type") == "content_block_start" and
|
|
|
+ e.get("content_block", {}).get("type") == "tool_use"
|
|
|
+ for e in events):
|
|
|
+ # Find tool use block start
|
|
|
+ tool_starts = [e for e in events if
|
|
|
+ e.get("type") == "content_block_start" and
|
|
|
+ e.get("content_block", {}).get("type") == "tool_use"]
|
|
|
+
|
|
|
+ assert len(tool_starts) > 0, "Should have tool_use content_block_start"
|
|
|
+
|
|
|
+ # Check index is correct (should be 0 if no text, 1 if there's text)
|
|
|
+ tool_start = tool_starts[0]
|
|
|
+ assert "index" in tool_start
|
|
|
+ assert tool_start["content_block"]["type"] == "tool_use"
|
|
|
+ assert "name" in tool_start["content_block"]
|
|
|
+
|
|
|
+
|
|
|
+# Vision/multimodal tests
|
|
|
+
|
|
|
+def test_anthropic_vision_format_accepted():
|
|
|
+ """Test that Anthropic vision format is accepted (format validation only)"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ # Small 1x1 red PNG image in base64
|
|
|
+ red_pixel_png = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg=="
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 10,
|
|
|
+ "messages": [
|
|
|
+ {
|
|
|
+ "role": "user",
|
|
|
+ "content": [
|
|
|
+ {
|
|
|
+ "type": "image",
|
|
|
+ "source": {
|
|
|
+ "type": "base64",
|
|
|
+ "media_type": "image/png",
|
|
|
+ "data": red_pixel_png
|
|
|
+ }
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "type": "text",
|
|
|
+ "text": "What is this?"
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ # Server accepts the format but tinyllama doesn't support images
|
|
|
+ # So it should return 500 with clear error message about missing mmproj
|
|
|
+ assert res.status_code == 500
|
|
|
+ assert "image input is not supported" in res.body.get("error", {}).get("message", "").lower()
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_vision_base64_with_multimodal_model(vision_server):
|
|
|
+ """Test vision with base64 image using Anthropic format with multimodal model"""
|
|
|
+ global server
|
|
|
+ server = vision_server
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ # Get test image in base64 format
|
|
|
+ image_base64 = get_test_image_base64()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 10,
|
|
|
+ "messages": [
|
|
|
+ {
|
|
|
+ "role": "user",
|
|
|
+ "content": [
|
|
|
+ {
|
|
|
+ "type": "image",
|
|
|
+ "source": {
|
|
|
+ "type": "base64",
|
|
|
+ "media_type": "image/png",
|
|
|
+ "data": image_base64
|
|
|
+ }
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "type": "text",
|
|
|
+ "text": "What is this:\n"
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ }
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200, f"Expected 200, got {res.status_code}: {res.body}"
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+ assert len(res.body["content"]) > 0
|
|
|
+ assert res.body["content"][0]["type"] == "text"
|
|
|
+ # The model should generate some response about the image
|
|
|
+ assert len(res.body["content"][0]["text"]) > 0
|
|
|
+
|
|
|
+
|
|
|
+# Parameter tests
|
|
|
+
|
|
|
+def test_anthropic_stop_sequences():
|
|
|
+ """Test stop_sequences parameter"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 100,
|
|
|
+ "stop_sequences": ["\n", "END"],
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Count to 10"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_temperature():
|
|
|
+ """Test temperature parameter"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 50,
|
|
|
+ "temperature": 0.5,
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Hello"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_top_p():
|
|
|
+ """Test top_p parameter"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 50,
|
|
|
+ "top_p": 0.9,
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Hello"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_top_k():
|
|
|
+ """Test top_k parameter (llama.cpp specific)"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 50,
|
|
|
+ "top_k": 40,
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Hello"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+
|
|
|
+
|
|
|
+# Error handling tests
|
|
|
+
|
|
|
+def test_anthropic_missing_messages():
|
|
|
+ """Test error when messages are missing"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 50
|
|
|
+ # missing "messages" field
|
|
|
+ })
|
|
|
+
|
|
|
+ # Should return an error (400 or 500)
|
|
|
+ assert res.status_code >= 400
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_empty_messages():
|
|
|
+ """Test permissive handling of empty messages array"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 50,
|
|
|
+ "messages": []
|
|
|
+ })
|
|
|
+
|
|
|
+ # Server is permissive and accepts empty messages (provides defaults)
|
|
|
+ # This matches the permissive validation design choice
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+
|
|
|
+
|
|
|
+# Content block index tests
|
|
|
+
|
|
|
+def test_anthropic_streaming_content_block_indices():
|
|
|
+ """Test that content block indices are correct in streaming"""
|
|
|
+ server.jinja = True
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ # Request that might produce both text and tool use
|
|
|
+ res = server.make_stream_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 200,
|
|
|
+ "stream": True,
|
|
|
+ "tools": [{
|
|
|
+ "name": "test_tool",
|
|
|
+ "description": "A test tool",
|
|
|
+ "input_schema": {
|
|
|
+ "type": "object",
|
|
|
+ "properties": {
|
|
|
+ "param": {"type": "string"}
|
|
|
+ },
|
|
|
+ "required": ["param"]
|
|
|
+ }
|
|
|
+ }],
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Use the test tool"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ events = []
|
|
|
+ for data in res:
|
|
|
+ events.append(data)
|
|
|
+
|
|
|
+ # Check content_block_start events have sequential indices
|
|
|
+ block_starts = [e for e in events if e.get("type") == "content_block_start"]
|
|
|
+ if len(block_starts) > 1:
|
|
|
+ # If there are multiple blocks, indices should be sequential
|
|
|
+ indices = [e["index"] for e in block_starts]
|
|
|
+ expected_indices = list(range(len(block_starts)))
|
|
|
+ assert indices == expected_indices, f"Expected indices {expected_indices}, got {indices}"
|
|
|
+
|
|
|
+ # Check content_block_stop events match the starts
|
|
|
+ block_stops = [e for e in events if e.get("type") == "content_block_stop"]
|
|
|
+ start_indices = set(e["index"] for e in block_starts)
|
|
|
+ stop_indices = set(e["index"] for e in block_stops)
|
|
|
+ assert start_indices == stop_indices, "content_block_stop indices should match content_block_start indices"
|
|
|
+
|
|
|
+
|
|
|
+# Extended features tests
|
|
|
+
|
|
|
+def test_anthropic_thinking():
|
|
|
+ """Test extended thinking parameter"""
|
|
|
+ server.jinja = True
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 100,
|
|
|
+ "thinking": {
|
|
|
+ "type": "enabled",
|
|
|
+ "budget_tokens": 50
|
|
|
+ },
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "What is 2+2?"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+
|
|
|
+
|
|
|
+def test_anthropic_metadata():
|
|
|
+ """Test metadata parameter"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 50,
|
|
|
+ "metadata": {
|
|
|
+ "user_id": "test_user_123"
|
|
|
+ },
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Hello"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert res.status_code == 200
|
|
|
+ assert res.body["type"] == "message"
|
|
|
+
|
|
|
+
|
|
|
+# Compatibility tests
|
|
|
+
|
|
|
+def test_anthropic_vs_openai_different_response_format():
|
|
|
+ """Verify Anthropic format is different from OpenAI format"""
|
|
|
+ server.start()
|
|
|
+
|
|
|
+ # Make OpenAI request
|
|
|
+ openai_res = server.make_request("POST", "/v1/chat/completions", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 50,
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Hello"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ # Make Anthropic request
|
|
|
+ anthropic_res = server.make_request("POST", "/v1/messages", data={
|
|
|
+ "model": "test",
|
|
|
+ "max_tokens": 50,
|
|
|
+ "messages": [
|
|
|
+ {"role": "user", "content": "Hello"}
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ assert openai_res.status_code == 200
|
|
|
+ assert anthropic_res.status_code == 200
|
|
|
+
|
|
|
+ # OpenAI has "object", Anthropic has "type"
|
|
|
+ assert "object" in openai_res.body
|
|
|
+ assert "type" in anthropic_res.body
|
|
|
+ assert openai_res.body["object"] == "chat.completion"
|
|
|
+ assert anthropic_res.body["type"] == "message"
|
|
|
+
|
|
|
+ # OpenAI has "choices", Anthropic has "content"
|
|
|
+ assert "choices" in openai_res.body
|
|
|
+ assert "content" in anthropic_res.body
|
|
|
+
|
|
|
+ # Different usage field names
|
|
|
+ assert "prompt_tokens" in openai_res.body["usage"]
|
|
|
+ assert "input_tokens" in anthropic_res.body["usage"]
|
|
|
+ assert "completion_tokens" in openai_res.body["usage"]
|
|
|
+ assert "output_tokens" in anthropic_res.body["usage"]
|
Fredrik Hultin