mmc/lingma-openai-gateway
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: lock Anthropic contract regressions

Browse Source 
Align TOOL_FORWARD_ENABLED docs with the current default and add count_tokens/auth/backpressure regressions so Anthropic compatibility stays stable.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
GitHub Actions
2026-04-22 13:03:25 +08:00
parent 8b012310a2
commit f7fad97073
3 changed files with 100 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
tests/test_tool_call_bridge.py
View File

@@ -1197,6 +1197,104 @@ class ToolCallBridgeTests(unittest.IsolatedAsyncioTestCase):
self.assertEqual(fake_cache.keys, [])
self.assertEqual(fake_cache.get_calls, [])
self.assertEqual(fake_cache.put_calls, [])
async def test_anthropic_count_tokens_returns_input_tokens(self) -> None:
req = AnthropicMessagesRequest(
model="claude-3-5-sonnet-20241022",
max_tokens=64,
messages=[{"role": "user", "content": "count me"}],
)
with patch.object(main.settings, "api_keys", ["test-key"]):
response = await main.v1_messages_count_tokens(
req,
_make_request(
"/v1/messages/count_tokens",
headers={"x-api-key": "test-key", "anthropic-version": "2023-06-01"},
),
)
payload = json.loads(response.body)
self.assertEqual(response.status_code, 200)
self.assertEqual(payload, {"input_tokens": main.estimate_tokens(main._messages_to_prompt(main.anthropic_to_internal_messages(req)))})
async def test_anthropic_count_tokens_requires_authentication(self) -> None:
req = AnthropicMessagesRequest(
model="claude-3-5-sonnet-20241022",
max_tokens=64,
messages=[{"role": "user", "content": "count me"}],
)
with patch.object(main.settings, "api_keys", ["test-key"]):
response = await main.v1_messages_count_tokens(
req,
_make_request(
"/v1/messages/count_tokens",
headers={"anthropic-version": "2023-06-01"},
),
)
payload = json.loads(response.body)
self.assertEqual(response.status_code, 401)
self.assertEqual(payload["type"], "error")
self.assertEqual(payload["error"]["type"], "authentication_error")
async def test_anthropic_messages_requires_authentication(self) -> None:
req = AnthropicMessagesRequest(
model="claude-3-5-sonnet-20241022",
max_tokens=64,
messages=[{"role": "user", "content": "hi"}],
stream=False,
)
with patch.object(main.settings, "api_keys", ["test-key"]):
response = await main.v1_messages(
req,
_make_request(
"/v1/messages",
headers={"anthropic-version": "2023-06-01"},
),
)
payload = json.loads(response.body)
self.assertEqual(response.status_code, 401)
self.assertEqual(payload["type"], "error")
self.assertEqual(payload["error"]["type"], "authentication_error")
async def test_anthropic_messages_backpressure_returns_overloaded_error(self) -> None:
fake_client = _FakeClient(stream_events=[], complete_result={"text": "ok", "toolEvents": []})
req = AnthropicMessagesRequest(
model="claude-3-5-sonnet-20241022",
max_tokens=64,
messages=[{"role": "user", "content": "hi"}],
stream=False,
)
fake_guard = types.SimpleNamespace(
in_flight=0,
try_acquire=AsyncMock(side_effect=main.BackpressureRejected(2.4)),
)
with (
patch.object(main, "pool", _FakePool(_FakeInstance(fake_client))),
patch.object(main, "chat_guard", fake_guard),
patch.object(main, "_ensure_instance_logged_in", AsyncMock(return_value={"id": "u"})),
patch.object(main.settings, "api_keys", ["test-key"]),
):
response = await main.v1_messages(
req,
_make_request(
"/v1/messages",
headers={"x-api-key": "test-key", "anthropic-version": "2023-06-01"},
),
)
payload = json.loads(response.body)
self.assertEqual(response.status_code, 429)
self.assertEqual(response.headers["Retry-After"], "2")
self.assertEqual(payload["type"], "error")
self.assertEqual(payload["error"]["type"], "overloaded_error")
self.assertIn("retry later", payload["error"]["message"])
async def test_responses_non_stream_maps_chat_payload_shape_and_input(self) -> None:
req = ResponsesRequest(
model="org_auto",