477 lines
24 KiB
Python
477 lines
24 KiB
Python
"""Tests for the Anthropic Messages API support (api/messages.py + requests/anthropic.py).
|
||
|
||
Covers the pure translation layer, the translated (Ollama-style) and native
|
||
(configured-anthropic-endpoint) backend paths, thinking→reasoning mapping,
|
||
streaming event shape, token counting, and the nomyo-cache reflection.
|
||
"""
|
||
from contextlib import ExitStack, contextmanager
|
||
from types import SimpleNamespace as NS
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import orjson
|
||
import pytest
|
||
|
||
import router
|
||
from api import messages as api_messages
|
||
from requests import anthropic as at
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Pure translation unit tests (no app / no I/O)
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestRequestTranslation:
|
||
def test_system_string(self):
|
||
chat = at.anthropic_messages_to_chat("be brief", [{"role": "user", "content": "hi"}])
|
||
assert chat[0] == {"role": "system", "content": "be brief"}
|
||
assert chat[1] == {"role": "user", "content": "hi"}
|
||
|
||
def test_system_blocks_join(self):
|
||
chat = at.anthropic_messages_to_chat(
|
||
[{"type": "text", "text": "a"}, {"type": "text", "text": "b"}],
|
||
[{"role": "user", "content": "hi"}])
|
||
assert chat[0] == {"role": "system", "content": "a\n\nb"}
|
||
|
||
def test_image_base64_to_data_url(self):
|
||
chat = at.anthropic_messages_to_chat(None, [{
|
||
"role": "user", "content": [
|
||
{"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": "AAA"}},
|
||
{"type": "text", "text": "what?"},
|
||
]}])
|
||
parts = chat[0]["content"]
|
||
assert parts[0] == {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAA"}}
|
||
assert parts[1] == {"type": "text", "text": "what?"}
|
||
|
||
def test_tool_use_and_result_roundtrip(self):
|
||
chat = at.anthropic_messages_to_chat(None, [
|
||
{"role": "assistant", "content": [
|
||
{"type": "text", "text": "calling"},
|
||
{"type": "tool_use", "id": "toolu_1", "name": "get", "input": {"x": 1}}]},
|
||
{"role": "user", "content": [
|
||
{"type": "tool_result", "tool_use_id": "toolu_1", "content": "42"}]},
|
||
])
|
||
assert chat[0]["role"] == "assistant"
|
||
assert chat[0]["content"] == "calling"
|
||
assert chat[0]["tool_calls"][0]["id"] == "toolu_1"
|
||
assert chat[0]["tool_calls"][0]["function"]["arguments"] == '{"x":1}'
|
||
assert chat[1] == {"role": "tool", "tool_call_id": "toolu_1", "content": "42"}
|
||
|
||
def test_tool_result_blocks_flattened(self):
|
||
chat = at.anthropic_messages_to_chat(None, [{
|
||
"role": "user", "content": [
|
||
{"type": "tool_result", "tool_use_id": "t1",
|
||
"content": [{"type": "text", "text": "line"}]}]}])
|
||
assert chat[0] == {"role": "tool", "tool_call_id": "t1", "content": "line"}
|
||
|
||
def test_tools_and_choice(self):
|
||
sp = at.anthropic_to_chat_send_params(
|
||
{"max_tokens": 10, "tools": [{"name": "get", "description": "d",
|
||
"input_schema": {"type": "object"}}],
|
||
"tool_choice": {"type": "tool", "name": "get"}},
|
||
[], "m")
|
||
assert sp["tools"] == [{"type": "function", "function": {
|
||
"name": "get", "description": "d", "parameters": {"type": "object"}}}]
|
||
assert sp["tool_choice"] == {"type": "function", "function": {"name": "get"}}
|
||
|
||
def test_tool_choice_variants(self):
|
||
assert at.tool_choice_anthropic_to_chat({"type": "auto"}) == "auto"
|
||
assert at.tool_choice_anthropic_to_chat({"type": "any"}) == "required"
|
||
assert at.tool_choice_anthropic_to_chat({"type": "none"}) == "none"
|
||
|
||
def test_server_tools_dropped(self):
|
||
# web_search has a type but no input_schema → not runnable locally
|
||
assert at.tools_anthropic_to_chat([{"type": "web_search_20260209", "name": "web_search"}]) is None
|
||
|
||
def test_stop_sequences_and_sampling(self):
|
||
sp = at.anthropic_to_chat_send_params(
|
||
{"max_tokens": 5, "stop_sequences": ["X"], "temperature": 0.2, "top_p": 0.9}, [], "m")
|
||
assert sp["stop"] == ["X"]
|
||
assert sp["temperature"] == 0.2
|
||
assert sp["top_p"] == 0.9
|
||
|
||
def test_thinking_maps_to_reasoning_effort(self):
|
||
assert at.anthropic_to_chat_send_params(
|
||
{"max_tokens": 5, "thinking": {"type": "enabled", "budget_tokens": 1000}}, [], "m"
|
||
)["reasoning_effort"] == "low"
|
||
assert at.anthropic_to_chat_send_params(
|
||
{"max_tokens": 5, "thinking": {"type": "enabled", "budget_tokens": 4096}}, [], "m"
|
||
)["reasoning_effort"] == "medium"
|
||
assert at.anthropic_to_chat_send_params(
|
||
{"max_tokens": 5, "thinking": {"type": "enabled", "budget_tokens": 20000}}, [], "m"
|
||
)["reasoning_effort"] == "high"
|
||
assert "reasoning_effort" not in at.anthropic_to_chat_send_params(
|
||
{"max_tokens": 5, "thinking": {"type": "disabled"}}, [], "m")
|
||
|
||
|
||
class TestResponseTranslation:
|
||
def test_content_blocks_order(self):
|
||
blocks = at.chat_message_to_content_blocks({
|
||
"role": "assistant", "reasoning_content": "hmm", "content": "answer",
|
||
"tool_calls": [{"id": "c1", "function": {"name": "f", "arguments": '{"a":1}'}}]})
|
||
assert [b["type"] for b in blocks] == ["thinking", "text", "tool_use"]
|
||
assert blocks[0]["thinking"] == "hmm"
|
||
assert blocks[1]["text"] == "answer"
|
||
assert blocks[2] == {"type": "tool_use", "id": "c1", "name": "f", "input": {"a": 1}}
|
||
|
||
def test_stop_reason_mapping(self):
|
||
assert at.finish_reason_to_stop_reason("stop") == "end_turn"
|
||
assert at.finish_reason_to_stop_reason("length") == "max_tokens"
|
||
assert at.finish_reason_to_stop_reason("tool_calls") == "tool_use"
|
||
assert at.finish_reason_to_stop_reason("stop", has_tool_use=True) == "tool_use"
|
||
|
||
def test_usage_mapping_no_cache(self):
|
||
u = at.usage_chat_to_anthropic({"prompt_tokens": 7, "completion_tokens": 3})
|
||
assert u == {"input_tokens": 7, "output_tokens": 3,
|
||
"cache_creation_input_tokens": 0, "cache_read_input_tokens": 0}
|
||
|
||
def test_usage_mapping_cached_tokens_subtracted(self):
|
||
# Backend reports 5 of 7 prompt tokens served from its prefix cache.
|
||
u = at.usage_chat_to_anthropic({
|
||
"prompt_tokens": 7, "completion_tokens": 3,
|
||
"prompt_tokens_details": {"cached_tokens": 5}})
|
||
assert u == {"input_tokens": 2, "output_tokens": 3,
|
||
"cache_creation_input_tokens": 0, "cache_read_input_tokens": 5}
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Streaming translator (pure)
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
def _chunk(content=None, tool_calls=None, reasoning=None, finish_reason=None):
|
||
delta = NS(content=content, tool_calls=tool_calls)
|
||
if reasoning is not None:
|
||
delta.reasoning_content = reasoning
|
||
return NS(choices=[NS(delta=delta, finish_reason=finish_reason)], usage=None)
|
||
|
||
|
||
def _usage_chunk(p, c, cached=None):
|
||
usage = NS(prompt_tokens=p, completion_tokens=c)
|
||
if cached is not None:
|
||
usage.prompt_tokens_details = NS(cached_tokens=cached)
|
||
return NS(choices=[], usage=usage)
|
||
|
||
|
||
async def _collect(translator, gen):
|
||
frames = []
|
||
async for sse in translator.events(gen):
|
||
frames.append(sse.decode())
|
||
return _parse_sse("".join(frames))
|
||
|
||
|
||
def _parse_sse(text):
|
||
out = []
|
||
for frame in text.strip().split("\n\n"):
|
||
if not frame.strip():
|
||
continue
|
||
etype = data = None
|
||
for line in frame.splitlines():
|
||
if line.startswith("event: "):
|
||
etype = line[len("event: "):]
|
||
elif line.startswith("data: "):
|
||
data = orjson.loads(line[len("data: "):])
|
||
out.append((etype, data))
|
||
return out
|
||
|
||
|
||
class TestStreamTranslator:
|
||
async def test_text_stream(self):
|
||
async def gen():
|
||
yield _chunk(content="Hel")
|
||
yield _chunk(content="lo", finish_reason="stop")
|
||
yield _usage_chunk(3, 5)
|
||
tr = at.ChatToMessagesStream("msg_1", "m")
|
||
events = await _collect(tr, gen())
|
||
types = [e[0] for e in events]
|
||
assert types[0] == "message_start"
|
||
assert types[-1] == "message_stop"
|
||
assert "content_block_start" in types and "content_block_stop" in types
|
||
text = "".join(d["delta"]["text"] for t, d in events
|
||
if t == "content_block_delta" and d["delta"]["type"] == "text_delta")
|
||
assert text == "Hello"
|
||
md = [d for t, d in events if t == "message_delta"][0]
|
||
assert md["delta"]["stop_reason"] == "end_turn"
|
||
assert md["usage"]["output_tokens"] == 5
|
||
assert tr.content_blocks == [{"type": "text", "text": "Hello"}]
|
||
|
||
async def test_cached_tokens_surface_as_cache_read(self):
|
||
async def gen():
|
||
yield _chunk(content="hi", finish_reason="stop")
|
||
yield _usage_chunk(10, 2, cached=6)
|
||
tr = at.ChatToMessagesStream("msg_1", "m")
|
||
events = await _collect(tr, gen())
|
||
md = [d for t, d in events if t == "message_delta"][0]
|
||
assert md["usage"]["cache_read_input_tokens"] == 6
|
||
assert md["usage"]["input_tokens"] == 4 # 10 prompt − 6 cached
|
||
assert md["usage"]["cache_creation_input_tokens"] == 0
|
||
|
||
async def test_thinking_then_text(self):
|
||
async def gen():
|
||
yield _chunk(reasoning="think ")
|
||
yield _chunk(reasoning="more")
|
||
yield _chunk(content="answer", finish_reason="stop")
|
||
tr = at.ChatToMessagesStream("msg_1", "m")
|
||
events = await _collect(tr, gen())
|
||
# thinking block (index 0) then text block (index 1)
|
||
starts = [(d["index"], d["content_block"]["type"])
|
||
for t, d in events if t == "content_block_start"]
|
||
assert starts == [(0, "thinking"), (1, "text")]
|
||
think = "".join(d["delta"]["thinking"] for t, d in events
|
||
if t == "content_block_delta" and d["delta"]["type"] == "thinking_delta")
|
||
assert think == "think more"
|
||
assert tr.content_blocks[0] == {"type": "thinking", "thinking": "think more"}
|
||
|
||
async def test_tool_call_stream(self):
|
||
tc0 = NS(index=0, id="call_1", function=NS(name="lookup", arguments='{"q":'))
|
||
tc1 = NS(index=0, id=None, function=NS(name=None, arguments='"hi"}'))
|
||
|
||
async def gen():
|
||
yield _chunk(tool_calls=[tc0])
|
||
yield _chunk(tool_calls=[tc1], finish_reason="tool_calls")
|
||
yield _usage_chunk(4, 2)
|
||
tr = at.ChatToMessagesStream("msg_1", "m")
|
||
events = await _collect(tr, gen())
|
||
assert any(t == "content_block_start" and d["content_block"]["type"] == "tool_use"
|
||
for t, d in events)
|
||
partial = "".join(d["delta"]["partial_json"] for t, d in events
|
||
if t == "content_block_delta" and d["delta"]["type"] == "input_json_delta")
|
||
assert partial == '{"q":"hi"}'
|
||
md = [d for t, d in events if t == "message_delta"][0]
|
||
assert md["delta"]["stop_reason"] == "tool_use"
|
||
assert tr.content_blocks[-1] == {"type": "tool_use", "id": "call_1",
|
||
"name": "lookup", "input": {"q": "hi"}}
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Cache-hit replay
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestCacheReplay:
|
||
def test_message_object_to_sse_roundtrip(self):
|
||
msg = at.build_message_object(
|
||
message_id="msg_1", model="m",
|
||
content_blocks=[{"type": "text", "text": "hi"}],
|
||
stop_reason="end_turn",
|
||
usage=at.usage_chat_to_anthropic({"prompt_tokens": 2, "completion_tokens": 1}))
|
||
events = _parse_sse(at.message_object_to_sse(msg).decode())
|
||
types = [e[0] for e in events]
|
||
assert types[0] == "message_start"
|
||
assert types[-1] == "message_stop"
|
||
text = "".join(d["delta"]["text"] for t, d in events
|
||
if t == "content_block_delta" and d["delta"]["type"] == "text_delta")
|
||
assert text == "hi"
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Route-level tests
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
@contextmanager
|
||
def _enter(*cms):
|
||
with ExitStack() as stack:
|
||
for cm in cms:
|
||
stack.enter_context(cm)
|
||
yield
|
||
|
||
|
||
def _fake_completion(content="hello", usage=(3, 5), reasoning=None, tool_calls=None,
|
||
finish_reason="stop", cached=None):
|
||
md = {"role": "assistant", "content": content}
|
||
if reasoning is not None:
|
||
md["reasoning_content"] = reasoning
|
||
if tool_calls is not None:
|
||
md["tool_calls"] = tool_calls
|
||
msg = MagicMock()
|
||
msg.model_dump.return_value = md
|
||
usage_dump = {"prompt_tokens": usage[0], "completion_tokens": usage[1],
|
||
"total_tokens": sum(usage)}
|
||
if cached is not None:
|
||
usage_dump["prompt_tokens_details"] = {"cached_tokens": cached}
|
||
usage_obj = MagicMock()
|
||
usage_obj.model_dump.return_value = usage_dump
|
||
return NS(choices=[NS(message=msg, finish_reason=finish_reason)], usage=usage_obj)
|
||
|
||
|
||
def _patch_backend(native=False, endpoint="http://ollama:11434", cache=None):
|
||
return (
|
||
patch.object(api_messages, "choose_endpoint",
|
||
AsyncMock(return_value=(endpoint, "test-model:latest"))),
|
||
patch.object(api_messages, "decrement_usage", AsyncMock()),
|
||
patch.object(api_messages, "is_anthropic_endpoint", return_value=native),
|
||
patch.object(api_messages, "_make_openai_client", return_value=MagicMock()),
|
||
patch.object(api_messages, "get_llm_cache", return_value=cache),
|
||
)
|
||
|
||
|
||
class TestTranslatedRoute:
|
||
async def test_nonstream(self, client):
|
||
with _enter(*_patch_backend(native=False),
|
||
patch.object(api_messages, "create_chat_with_retries",
|
||
AsyncMock(return_value=_fake_completion("hello world")))):
|
||
resp = await client.post("/v1/messages",
|
||
json={"model": "test-model", "max_tokens": 100,
|
||
"messages": [{"role": "user", "content": "hi"}]})
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["type"] == "message"
|
||
assert body["role"] == "assistant"
|
||
assert body["content"] == [{"type": "text", "text": "hello world"}]
|
||
assert body["stop_reason"] == "end_turn"
|
||
assert body["usage"]["input_tokens"] == 3 and body["usage"]["output_tokens"] == 5
|
||
assert body["id"].startswith("msg_")
|
||
|
||
async def test_nonstream_cached_tokens(self, client):
|
||
with _enter(*_patch_backend(native=False),
|
||
patch.object(api_messages, "create_chat_with_retries",
|
||
AsyncMock(return_value=_fake_completion(
|
||
"hi", usage=(10, 4), cached=6)))):
|
||
resp = await client.post("/v1/messages",
|
||
json={"model": "test-model", "max_tokens": 100,
|
||
"messages": [{"role": "user", "content": "hi"}]})
|
||
u = resp.json()["usage"]
|
||
assert u["cache_read_input_tokens"] == 6
|
||
assert u["input_tokens"] == 4
|
||
assert u["cache_creation_input_tokens"] == 0
|
||
|
||
async def test_missing_max_tokens_400(self, client):
|
||
with _enter(*_patch_backend(native=False)):
|
||
resp = await client.post("/v1/messages",
|
||
json={"model": "m", "messages": [{"role": "user", "content": "hi"}]})
|
||
assert resp.status_code == 400
|
||
|
||
async def test_nonstream_tool_use(self, client):
|
||
tc = [{"id": "c1", "function": {"name": "get", "arguments": '{"a":1}'}}]
|
||
with _enter(*_patch_backend(native=False),
|
||
patch.object(api_messages, "create_chat_with_retries",
|
||
AsyncMock(return_value=_fake_completion(
|
||
content=None, tool_calls=tc, finish_reason="tool_calls")))):
|
||
resp = await client.post("/v1/messages",
|
||
json={"model": "m", "max_tokens": 50,
|
||
"messages": [{"role": "user", "content": "call it"}]})
|
||
body = resp.json()
|
||
assert body["stop_reason"] == "tool_use"
|
||
assert body["content"][0] == {"type": "tool_use", "id": "c1", "name": "get", "input": {"a": 1}}
|
||
|
||
async def test_stream_event_sequence(self, client):
|
||
async def _text():
|
||
yield _chunk(content="Hi", finish_reason="stop")
|
||
yield _usage_chunk(3, 2)
|
||
with _enter(*_patch_backend(native=False),
|
||
patch.object(api_messages, "create_chat_with_retries",
|
||
AsyncMock(return_value=_text()))):
|
||
resp = await client.post("/v1/messages",
|
||
json={"model": "m", "max_tokens": 50, "stream": True,
|
||
"messages": [{"role": "user", "content": "hi"}]})
|
||
assert resp.headers["content-type"].startswith("text/event-stream")
|
||
events = _parse_sse(resp.content.decode())
|
||
types = [e[0] for e in events]
|
||
assert types[0] == "message_start" and types[-1] == "message_stop"
|
||
text = "".join(d["delta"]["text"] for t, d in events
|
||
if t == "content_block_delta" and d["delta"]["type"] == "text_delta")
|
||
assert text == "Hi"
|
||
|
||
async def test_thinking_passed_as_reasoning_effort(self, client):
|
||
captured = {}
|
||
|
||
async def _spy(oclient, send_params, endpoint, model, tracking_model):
|
||
captured.update(send_params)
|
||
return _fake_completion("ok")
|
||
|
||
with _enter(*_patch_backend(native=False),
|
||
patch.object(api_messages, "create_chat_with_retries", _spy)):
|
||
await client.post("/v1/messages",
|
||
json={"model": "m", "max_tokens": 50,
|
||
"thinking": {"type": "enabled", "budget_tokens": 20000},
|
||
"messages": [{"role": "user", "content": "hi"}]})
|
||
assert captured["reasoning_effort"] == "high"
|
||
|
||
|
||
class TestCacheRoute:
|
||
async def test_hit_reports_cache_read(self, client):
|
||
stored = at.build_message_object(
|
||
message_id="msg_old", model="m",
|
||
content_blocks=[{"type": "text", "text": "cached"}], stop_reason="end_turn",
|
||
usage=at.usage_chat_to_anthropic({"prompt_tokens": 9, "completion_tokens": 4}))
|
||
fake_cache = MagicMock()
|
||
fake_cache.get_chat = AsyncMock(return_value=orjson.dumps(stored))
|
||
with _enter(*_patch_backend(native=False, cache=fake_cache)):
|
||
resp = await client.post("/v1/messages",
|
||
json={"model": "m", "max_tokens": 50, "nomyo": {"cache": True},
|
||
"messages": [{"role": "user", "content": "hi"}]})
|
||
body = resp.json()
|
||
assert body["content"] == [{"type": "text", "text": "cached"}]
|
||
assert body["usage"]["cache_read_input_tokens"] == 9
|
||
assert body["usage"]["input_tokens"] == 0
|
||
assert body["id"].startswith("msg_") and body["id"] != "msg_old"
|
||
|
||
|
||
class TestNativeRoute:
|
||
def _fake_client(self, *, post_return=None, stream_frames=None, status=200):
|
||
client = MagicMock()
|
||
if post_return is not None:
|
||
resp = MagicMock()
|
||
resp.status_code = status
|
||
resp.json.return_value = post_return
|
||
client.post = AsyncMock(return_value=resp)
|
||
if stream_frames is not None:
|
||
class _Stream:
|
||
async def __aenter__(self_):
|
||
return self_
|
||
async def __aexit__(self_, *a):
|
||
return False
|
||
async def aiter_bytes(self_):
|
||
for f in stream_frames:
|
||
yield f
|
||
client.stream = MagicMock(return_value=_Stream())
|
||
return client
|
||
|
||
async def test_nonstream_passthrough(self, client):
|
||
upstream = {"id": "msg_upstream", "type": "message", "role": "assistant",
|
||
"content": [{"type": "text", "text": "native hi"}],
|
||
"stop_reason": "end_turn",
|
||
"usage": {"input_tokens": 2, "output_tokens": 3}}
|
||
fake = self._fake_client(post_return=upstream)
|
||
with _enter(*_patch_backend(native=True, endpoint="https://api.anthropic.com"),
|
||
patch.object(api_messages, "_anthropic_http_client", return_value=fake)):
|
||
resp = await client.post("/v1/messages",
|
||
json={"model": "claude-x", "max_tokens": 50,
|
||
"messages": [{"role": "user", "content": "hi"}]})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["content"][0]["text"] == "native hi"
|
||
# request forwarded verbatim with stream disabled, nomyo stripped
|
||
sent = fake.post.call_args.kwargs["json"]
|
||
assert sent["stream"] is False and "nomyo" not in sent
|
||
headers = fake.post.call_args.kwargs["headers"]
|
||
assert headers["x-api-key"] and headers["anthropic-version"]
|
||
|
||
async def test_stream_passthrough(self, client):
|
||
frames = [
|
||
b'event: message_start\ndata: {"type":"message_start","message":{"usage":{"input_tokens":5}}}\n\n',
|
||
b'event: message_delta\ndata: {"type":"message_delta","usage":{"output_tokens":7}}\n\n',
|
||
b'event: message_stop\ndata: {"type":"message_stop"}\n\n',
|
||
]
|
||
fake = self._fake_client(stream_frames=frames)
|
||
tracked = []
|
||
with _enter(*_patch_backend(native=True, endpoint="https://api.anthropic.com"),
|
||
patch.object(api_messages, "_anthropic_http_client", return_value=fake),
|
||
patch.object(api_messages, "_track",
|
||
AsyncMock(side_effect=lambda *a: tracked.append(a)))):
|
||
resp = await client.post("/v1/messages",
|
||
json={"model": "claude-x", "max_tokens": 50, "stream": True,
|
||
"messages": [{"role": "user", "content": "hi"}]})
|
||
body = resp.content.decode()
|
||
assert "message_start" in body and "message_stop" in body
|
||
# usage parsed out of the proxied stream for token tracking
|
||
assert tracked and tracked[0][2] == 5 and tracked[0][3] == 7
|
||
|
||
|
||
class TestCountTokens:
|
||
async def test_local_estimate(self, client):
|
||
with _enter(patch.object(api_messages, "choose_endpoint",
|
||
AsyncMock(return_value=("http://ollama:11434", "m"))),
|
||
patch.object(api_messages, "is_anthropic_endpoint", return_value=False)):
|
||
resp = await client.post("/v1/messages/count_tokens",
|
||
json={"model": "m",
|
||
"messages": [{"role": "user", "content": "count me"}]})
|
||
assert resp.status_code == 200
|
||
assert isinstance(resp.json()["input_tokens"], int)
|
||
assert resp.json()["input_tokens"] > 0
|