scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-06 02:48:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103
litellm/tests/unit/llms/bedrock/test_mantle.py
yuneng-jiang 5e6dc89ba1
test: move tests/test_litellm/llms into tests/unit/llms (#43191) 
* ci: run the unit_selection.sh shard files on every event instead of only fork pull requests

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* ci: rename fork-flag to unit-flag now that it applies on every event

* test: move tests/test_litellm root and small trees into tests/unit

Pure renames, no content changes. Follow-up commits in this PR fix
references, merge the three files that already existed in tests/unit,
keep live-provider tests in tests/test_litellm and wire CI.

* test: carry tests/test_litellm conftest isolation into tests/unit

Callback lists, routing fallbacks, cached HTTP clients, logger state, AWS,
proxy-URL and keychain env, and session-end client cleanup now reset for
unit tests too. The environment isolation owns its MonkeyPatch so a test's
own monkeypatch is undone before the model-cost teardown runs.

* test: merge, split and prune the moved root and small-tree tests

Merge batches/test_batch_utils.py and the chat_completions and messages
dispatch tests into the files that already existed in tests/unit. Keep
the live Gemini interactions tests, the async image-fetch format test and
the OpenAI embedding scorer test in tests/test_litellm since they need
real network or keys. Put test_router.py under tests/unit/test_router so
the existing package no longer shadows it. Delete eight tests the audit
found superseded by stronger ones kept in this move.

* ci: run the moved root and small-tree tests under their legacy flags

Add the misc and responses-caching-types flags to unit_selection.sh and
CircleCI, extend enterprise-routing and mcp-integration, and point the
legacy GHA shards, Makefile, redis-compat workflow, merge smoke manifest
and change classifier at the new paths.

* test: make the new tests/unit directories packages

tests/unit/test_package_layout.py requires every directory to carry an
__init__.py, and without one the moved and retained
test_litellm_responses_bridge.py modules collide on import.

* test: scope the unit socket block to tests/unit in shared sessions

The GHA shards collect the legacy test-path and the unit selection in one
pytest session. The unit conftest's loopback-only block leaked into legacy
modules that reach the network at import. The legacy conftest now lifts the
restriction at collect and setup time, and the unit conftest re-applies it
when collecting its own modules.

* test: move tests/test_litellm/llms into tests/unit/llms

Rename-only. Moves the provider tests and the fine-tuning fixtures they
load, mirroring the old paths. Follow-up commits merge, split and wire them.

* test: merge, split and prune the moved llms tests

Merges the Databricks chat transformation tests into the existing unit
file, keeps the tests that need real keys or the network in
tests/test_litellm, deletes the audited tests a stronger unit test
already covers, and points imports at tests.unit.llms.

* ci: run the moved llms tests under their legacy flags

The Vertex AI and All Other Providers shards keep their legacy test-path
for the retained files and add the llm-vertex-ai and llm-other-providers
unit selections. CircleCI gets matching unit jobs.

* test: make the tests/unit/llms directories packages

Adds __init__.py to the moved dirs and drops the legacy ones whose
directories no longer hold tests.

* test: drop script runners and path hacks the llms split left dangling

The __main__ runners in the split openai_like files and the Databricks e2e
runner called tests that now live in the other half of the split or were
deleted. The retained legacy halves also no longer need sys.path edits.

* test: give the shard-script tests their own GITHUB_OUTPUT

They only passed where the runner set it. The CircleCI unit job's env
allowlist drops it, so the script's redirect failed there.

* test: point the router and module-deletion checks at tests/unit

router_code_coverage and code_qa_check_tests only searched tests/test_litellm,
so the moved router tests no longer counted. The two silent-experiment tests
the audit deleted were the only direct callers of those methods; they are
replaced with tests that assert the forwarded shadow request and the
recursion guard.

* test: keep the Databricks manual e2e runner and fix the SageMaker Nova run path

The Databricks e2e file is a manual script whose main() calls the tests
that were pruned, so pruning them broke the documented run. It is back to
its main version. The SageMaker Nova docstring now points at the file's
real location in tests/local_testing.

* test: keep the job's UNIT_FLAG out of the shard-script tests

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-09-25 12:43:23 -07:00

794 lines
27 KiB
Python
Raw Blame History

"""
Unit tests for the Bedrock Mantle (Claude Mythos Preview) integration.
Tests cover route detection, URL construction, config dispatch for both
the /chat/completions and /messages endpoints, and project (workspace)
association via `aws_bedrock_project_id`.
"""
import json
from unittest.mock import patch
import httpx
import pytest
from litellm.llms.bedrock.common_utils import BedrockModelInfo, get_bedrock_chat_config
from litellm.llms.bedrock.chat.mantle.transformation import AmazonMantleConfig
from litellm.llms.bedrock.messages.mantle_transformation import (
AmazonMantleMessagesConfig,
)
def _anthropic_response(url: str) -> httpx.Response:
return httpx.Response(
status_code=200,
json={
"id": "msg_test",
"type": "message",
"role": "assistant",
"model": "anthropic.claude-mythos-preview",
"content": [{"type": "text", "text": "ok"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 1, "output_tokens": 1},
},
request=httpx.Request("POST", url),
)
def _capture_request(url: str, headers: dict, data) -> dict:
raw_body = data.decode("utf-8") if isinstance(data, bytes) else data or "{}"
return {
"path": httpx.URL(url).path,
"headers": headers,
"body": json.loads(raw_body),
}
def test_get_bedrock_route_mantle():
assert (
BedrockModelInfo.get_bedrock_route("mantle/anthropic.claude-mythos-preview")
== "mantle"
)
def test_get_bedrock_route_mantle_does_not_match_other_routes():
assert (
BedrockModelInfo.get_bedrock_route("anthropic.claude-3-sonnet-20240229-v1:0")
!= "mantle"
)
assert (
BedrockModelInfo.get_bedrock_route("converse/anthropic.claude-3-sonnet")
!= "mantle"
)
def test_explicit_mantle_route_flag():
assert (
BedrockModelInfo._explicit_mantle_route(
"mantle/anthropic.claude-mythos-preview"
)
is True
)
assert BedrockModelInfo._explicit_mantle_route("anthropic.claude-3-sonnet") is False
assert (
BedrockModelInfo._explicit_mantle_route("converse/anthropic.claude-3-sonnet")
is False
)
def test_mantle_url_construction():
config = AmazonMantleConfig()
url = config.get_complete_url(
api_base=None,
api_key=None,
model="mantle/anthropic.claude-mythos-preview",
optional_params={"aws_region_name": "us-east-1"},
litellm_params={},
)
assert url == "https://bedrock-mantle.us-east-1.api.aws/anthropic/v1/messages"
def test_mantle_url_construction_different_region():
config = AmazonMantleConfig()
url = config.get_complete_url(
api_base=None,
api_key=None,
model="mantle/anthropic.claude-mythos-preview",
optional_params={"aws_region_name": "us-west-2"},
litellm_params={},
)
assert url == "https://bedrock-mantle.us-west-2.api.aws/anthropic/v1/messages"
def test_get_bedrock_chat_config_returns_mantle_config():
config = get_bedrock_chat_config("mantle/anthropic.claude-mythos-preview")
assert isinstance(config, AmazonMantleConfig)
def test_get_bedrock_provider_config_for_messages_api_mantle():
config = BedrockModelInfo.get_bedrock_provider_config_for_messages_api(
"mantle/anthropic.claude-mythos-preview"
)
assert isinstance(config, AmazonMantleMessagesConfig)
def test_mantle_messages_url_construction():
config = AmazonMantleMessagesConfig()
url = config.get_complete_url(
api_base=None,
api_key=None,
model="mantle/anthropic.claude-mythos-preview",
optional_params={"aws_region_name": "us-east-1"},
litellm_params={},
)
assert url == "https://bedrock-mantle.us-east-1.api.aws/anthropic/v1/messages"
_VPC_ENDPOINT = "https://vpce-0a1b2c3d.bedrock-mantle.us-gov-west-1.vpce.amazonaws.com"
@pytest.fixture(autouse=True)
def no_ambient_mantle_api_base(monkeypatch):
monkeypatch.delenv("BEDROCK_MANTLE_API_BASE", raising=False)
def test_mantle_chat_url_honors_api_base_host():
config = AmazonMantleConfig()
url = config.get_complete_url(
api_base=_VPC_ENDPOINT,
api_key=None,
model="mantle/anthropic.claude-mythos-preview",
optional_params={"aws_region_name": "us-gov-west-1"},
litellm_params={},
)
assert url == f"{_VPC_ENDPOINT}/anthropic/v1/messages"
def test_mantle_chat_url_honors_api_base_full_path_without_duplication():
config = AmazonMantleConfig()
full = f"{_VPC_ENDPOINT}/anthropic/v1/messages"
url = config.get_complete_url(
api_base=full,
api_key=None,
model="mantle/anthropic.claude-mythos-preview",
optional_params={"aws_region_name": "us-gov-west-1"},
litellm_params={},
)
assert url == full
def test_mantle_messages_url_honors_api_base_host():
config = AmazonMantleMessagesConfig()
url = config.get_complete_url(
api_base=_VPC_ENDPOINT,
api_key=None,
model="mantle/anthropic.claude-mythos-preview",
optional_params={"aws_region_name": "us-gov-west-1"},
litellm_params={},
)
assert url == f"{_VPC_ENDPOINT}/anthropic/v1/messages"
assert "api.aws" not in url
def test_mantle_messages_url_honors_api_base_with_trailing_slash():
config = AmazonMantleMessagesConfig()
url = config.get_complete_url(
api_base=f"{_VPC_ENDPOINT}/",
api_key=None,
model="mantle/anthropic.claude-mythos-preview",
optional_params={"aws_region_name": "us-gov-west-1"},
litellm_params={},
)
assert url == f"{_VPC_ENDPOINT}/anthropic/v1/messages"
def test_mantle_messages_url_honors_aws_bedrock_runtime_endpoint():
config = AmazonMantleMessagesConfig()
url = config.get_complete_url(
api_base=None,
api_key=None,
model="mantle/anthropic.claude-mythos-preview",
optional_params={
"aws_region_name": "us-gov-west-1",
"aws_bedrock_runtime_endpoint": _VPC_ENDPOINT,
},
litellm_params={},
)
assert url == f"{_VPC_ENDPOINT}/anthropic/v1/messages"
_ENV_ENDPOINT = "https://bedrock-mantle.us-east-1.api.aws.internal.example.com"
@pytest.mark.parametrize("config_cls", [AmazonMantleConfig, AmazonMantleMessagesConfig])
@pytest.mark.parametrize(
"env_value",
[_ENV_ENDPOINT, f"{_ENV_ENDPOINT}/", f"{_ENV_ENDPOINT}/v1", f"{_ENV_ENDPOINT}/openai/v1"],
)
def test_mantle_url_honors_bedrock_mantle_api_base_env(monkeypatch, config_cls, env_value):
monkeypatch.setenv("BEDROCK_MANTLE_API_BASE", env_value)
url = config_cls().get_complete_url(
api_base=None,
api_key=None,
model="mantle/anthropic.claude-mythos-preview",
optional_params={"aws_region_name": "us-east-1"},
litellm_params={},
)
assert url == f"{_ENV_ENDPOINT}/anthropic/v1/messages"
@pytest.mark.parametrize("config_cls", [AmazonMantleConfig, AmazonMantleMessagesConfig])
@pytest.mark.parametrize(
("api_base", "optional_params"),
[
(_VPC_ENDPOINT, {"aws_region_name": "us-gov-west-1"}),
(None, {"aws_region_name": "us-gov-west-1", "aws_bedrock_runtime_endpoint": _VPC_ENDPOINT}),
],
)
def test_mantle_url_explicit_endpoint_beats_bedrock_mantle_api_base_env(
monkeypatch, config_cls, api_base, optional_params
):
monkeypatch.setenv("BEDROCK_MANTLE_API_BASE", _ENV_ENDPOINT)
url = config_cls().get_complete_url(
api_base=api_base,
api_key=None,
model="mantle/anthropic.claude-mythos-preview",
optional_params=optional_params,
litellm_params={},
)
assert url == f"{_VPC_ENDPOINT}/anthropic/v1/messages"
def test_mantle_transform_request_strips_prefix_and_adds_model():
config = AmazonMantleConfig()
request = config.transform_request(
model="mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "Hello"}],
optional_params={"max_tokens": 100},
litellm_params={},
headers={},
)
assert request["model"] == "anthropic.claude-mythos-preview"
assert "mantle/" not in request["model"]
assert "stream" not in request
def test_mantle_transform_request_keeps_stream_in_body():
config = AmazonMantleConfig()
request = config.transform_request(
model="mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "Hello"}],
optional_params={"max_tokens": 100, "stream": True},
litellm_params={},
headers={},
)
assert request["stream"] is True
assert request["model"] == "anthropic.claude-mythos-preview"
@pytest.mark.asyncio
async def test_mantle_async_transform_request_keeps_stream_in_body():
config = AmazonMantleConfig()
request = await config.async_transform_request(
model="mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "Hello"}],
optional_params={"max_tokens": 100, "stream": True},
litellm_params={},
headers={},
)
assert request["stream"] is True
assert request["model"] == "anthropic.claude-mythos-preview"
@pytest.mark.asyncio
async def test_mantle_async_transform_request_omits_stream_when_not_streaming():
config = AmazonMantleConfig()
request = await config.async_transform_request(
model="mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "Hello"}],
optional_params={"max_tokens": 100},
litellm_params={},
headers={},
)
assert "stream" not in request
def test_mantle_messages_transform_request_keeps_stream_in_body():
from litellm.types.router import GenericLiteLLMParams
config = AmazonMantleMessagesConfig()
request = config.transform_anthropic_messages_request(
model="mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "Hello"}],
anthropic_messages_optional_request_params={"max_tokens": 100, "stream": True},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert request["stream"] is True
assert request["model"] == "anthropic.claude-mythos-preview"
def test_mantle_messages_transform_request_omits_stream_when_not_streaming():
from litellm.types.router import GenericLiteLLMParams
config = AmazonMantleMessagesConfig()
request = config.transform_anthropic_messages_request(
model="mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "Hello"}],
anthropic_messages_optional_request_params={"max_tokens": 100},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert "stream" not in request
def test_mantle_chat_streaming_uses_anthropic_sse_iterator():
from litellm.llms.anthropic.chat.handler import ModelResponseIterator
config = AmazonMantleConfig()
assert config.has_custom_stream_wrapper is False
iterator = config.get_model_response_iterator(
streaming_response=iter([]),
sync_stream=True,
)
assert isinstance(iterator, ModelResponseIterator)
def test_mantle_validate_environment_sets_workspace_header():
config = AmazonMantleConfig()
headers = config.validate_environment(
headers={},
model="mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "Hello"}],
optional_params={},
litellm_params={"aws_bedrock_project_id": "proj_abc123def456"},
)
assert headers["anthropic-workspace"] == "proj_abc123def456"
def test_mantle_validate_environment_without_project_id():
config = AmazonMantleConfig()
headers = config.validate_environment(
headers={},
model="mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "Hello"}],
optional_params={},
litellm_params={"aws_bedrock_project_id": None},
)
assert "anthropic-workspace" not in headers
def test_mantle_messages_validate_environment_sets_workspace_header():
config = AmazonMantleMessagesConfig()
headers, api_base = config.validate_anthropic_messages_environment(
headers={},
model="mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "Hello"}],
optional_params={},
litellm_params={"aws_bedrock_project_id": "proj_abc123def456"},
api_base="https://bedrock-mantle.us-east-1.api.aws/anthropic/v1/messages",
)
assert headers["anthropic-workspace"] == "proj_abc123def456"
assert api_base == "https://bedrock-mantle.us-east-1.api.aws/anthropic/v1/messages"
def test_mantle_messages_validate_environment_without_project_id():
config = AmazonMantleMessagesConfig()
headers, _ = config.validate_anthropic_messages_environment(
headers={},
model="mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "Hello"}],
optional_params={},
litellm_params={},
)
assert "anthropic-workspace" not in headers
def test_mantle_completion_sends_workspace_header_and_clean_body():
import litellm
requests = []
def mock_post(self, url, data=None, headers=None, **kwargs):
requests.append(_capture_request(url=url, headers=headers or {}, data=data))
return _anthropic_response(url)
with patch("litellm.llms.custom_httpx.http_handler.HTTPHandler.post", mock_post):
response = litellm.completion(
model="bedrock/mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "hello"}],
max_tokens=10,
aws_bedrock_project_id="proj_abc123def456",
aws_access_key_id="fake-key",
aws_secret_access_key="fake-secret",
aws_region_name="us-east-1",
)
assert response.choices[0].message.content == "ok"
assert len(requests) == 1
assert requests[0]["path"] == "/anthropic/v1/messages"
assert requests[0]["headers"]["anthropic-workspace"] == "proj_abc123def456"
assert "aws_bedrock_project_id" not in requests[0]["body"]
@pytest.mark.asyncio
async def test_mantle_anthropic_messages_sends_workspace_header_and_clean_body():
import litellm
requests = []
async def mock_post(self, url, data=None, headers=None, **kwargs):
requests.append(_capture_request(url=url, headers=headers or {}, data=data))
return _anthropic_response(url)
try:
with patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new=mock_post,
):
response = await litellm.anthropic_messages(
model="bedrock/mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "hello"}],
max_tokens=10,
aws_bedrock_project_id="proj_abc123def456",
aws_access_key_id="fake-key",
aws_secret_access_key="fake-secret",
aws_region_name="us-east-1",
)
finally:
await litellm.close_litellm_async_clients()
assert response["content"][0]["text"] == "ok"
assert len(requests) == 1
assert requests[0]["path"] == "/anthropic/v1/messages"
assert requests[0]["headers"]["anthropic-workspace"] == "proj_abc123def456"
assert "aws_bedrock_project_id" not in requests[0]["body"]
async def _send_anthropic_messages_with_betas(**request_params: object) -> dict:
import litellm
requests = []
async def mock_post(self, url, data=None, headers=None, **kwargs):
requests.append(_capture_request(url=url, headers=headers or {}, data=data))
return _anthropic_response(url)
try:
with patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new=mock_post,
):
await litellm.anthropic_messages(
model="bedrock/mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "hello"}],
max_tokens=10,
aws_access_key_id="fake-key",
aws_secret_access_key="fake-secret",
aws_region_name="us-east-1",
**request_params,
)
finally:
await litellm.close_litellm_async_clients()
assert len(requests) == 1
return requests[0]
@pytest.mark.asyncio
@pytest.mark.usefixtures("local_beta_headers_config")
async def test_mantle_anthropic_messages_sends_every_beta_in_the_header_not_the_body():
sent = await _send_anthropic_messages_with_betas(
extra_headers={"anthropic-beta": "context-1m-2025-08-07,interleaved-thinking-2025-05-14"},
context_management={"edits": [{"type": "clear_tool_uses_20250919"}]},
)
assert (
sent["headers"]["anthropic-beta"]
== "context-1m-2025-08-07,context-management-2025-06-27,interleaved-thinking-2025-05-14"
)
assert sent["headers"]["anthropic-version"] == "2023-06-01"
assert sent["body"]["context_management"] == {"edits": [{"type": "clear_tool_uses_20250919"}]}
assert "anthropic_beta" not in sent["body"]
assert "anthropic_version" not in sent["body"]
@pytest.mark.asyncio
@pytest.mark.usefixtures("local_beta_headers_config")
async def test_mantle_anthropic_messages_drops_the_beta_header_when_mantle_rejects_every_value():
sent = await _send_anthropic_messages_with_betas(extra_headers={"anthropic-beta": "code-execution-2025-08-25"})
assert "anthropic-beta" not in sent["headers"]
assert "anthropic_beta" not in sent["body"]
def _usageless_anthropic_response(url: str) -> httpx.Response:
return httpx.Response(
status_code=200,
json={
"id": "msg_classifier",
"type": "message",
"role": "assistant",
"model": "anthropic.claude-opus-4-8",
"content": [{"type": "text", "text": "safe"}],
"stop_reason": "end_turn",
"stop_sequence": None,
},
request=httpx.Request("POST", url),
)
@pytest.mark.asyncio
async def test_mantle_anthropic_messages_backfills_missing_usage():
"""
Regression for LIT-4758: a Mantle non-streaming response with no `usage`
object must not reach the client usage-less, or Claude Code's auto-mode
classifier crashes on `usage.input_tokens`.
"""
import litellm
async def mock_post(self, url, data=None, headers=None, **kwargs):
return _usageless_anthropic_response(str(url))
try:
with patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new=mock_post,
):
response = await litellm.anthropic_messages(
model="bedrock/mantle/anthropic.claude-opus-4-8",
messages=[{"role": "user", "content": "is `Bash(ls)` safe?"}],
max_tokens=10,
aws_access_key_id="fake-key",
aws_secret_access_key="fake-secret",
aws_region_name="us-east-1",
)
finally:
await litellm.close_litellm_async_clients()
assert response["usage"]["input_tokens"] == 0
assert response["usage"]["output_tokens"] == 0
@pytest.mark.asyncio
async def test_mantle_anthropic_messages_preserves_upstream_usage():
"""Backfill must not clobber a usage object the upstream did return."""
import litellm
def _response_with_usage(url: str) -> httpx.Response:
return httpx.Response(
status_code=200,
json={
"id": "msg_test",
"type": "message",
"role": "assistant",
"model": "anthropic.claude-opus-4-8",
"content": [{"type": "text", "text": "ok"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {
"input_tokens": 42,
"output_tokens": 7,
"cache_read_input_tokens": 5,
},
},
request=httpx.Request("POST", url),
)
async def mock_post(self, url, data=None, headers=None, **kwargs):
return _response_with_usage(str(url))
try:
with patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new=mock_post,
):
response = await litellm.anthropic_messages(
model="bedrock/mantle/anthropic.claude-opus-4-8",
messages=[{"role": "user", "content": "hello"}],
max_tokens=10,
aws_access_key_id="fake-key",
aws_secret_access_key="fake-secret",
aws_region_name="us-east-1",
)
finally:
await litellm.close_litellm_async_clients()
assert response["usage"]["input_tokens"] == 42
assert response["usage"]["output_tokens"] == 7
assert response["usage"]["cache_read_input_tokens"] == 5
@pytest.mark.asyncio
async def test_mantle_anthropic_messages_routes_to_vpc_api_base():
import litellm
urls = []
async def mock_post(self, url, data=None, headers=None, **kwargs):
urls.append(str(url))
return _anthropic_response(str(url))
try:
with patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new=mock_post,
):
await litellm.anthropic_messages(
model="bedrock/mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "hello"}],
max_tokens=10,
api_base=_VPC_ENDPOINT,
aws_access_key_id="fake-key",
aws_secret_access_key="fake-secret",
aws_region_name="us-gov-west-1",
)
finally:
await litellm.close_litellm_async_clients()
assert len(urls) == 1
assert urls[0] == f"{_VPC_ENDPOINT}/anthropic/v1/messages"
assert "api.aws" not in urls[0]
_ANTHROPIC_SSE_EVENTS = (
(
"message_start",
{
"type": "message_start",
"message": {
"id": "msg_stream_test",
"type": "message",
"role": "assistant",
"model": "anthropic.claude-mythos-preview",
"content": [],
"stop_reason": None,
"stop_sequence": None,
"usage": {"input_tokens": 10, "output_tokens": 1},
},
},
),
(
"content_block_start",
{
"type": "content_block_start",
"index": 0,
"content_block": {"type": "text", "text": ""},
},
),
(
"content_block_delta",
{
"type": "content_block_delta",
"index": 0,
"delta": {"type": "text_delta", "text": "pong"},
},
),
("content_block_stop", {"type": "content_block_stop", "index": 0}),
(
"message_delta",
{
"type": "message_delta",
"delta": {"stop_reason": "end_turn", "stop_sequence": None},
"usage": {"output_tokens": 2},
},
),
("message_stop", {"type": "message_stop"}),
)
def _anthropic_sse_bytes() -> bytes:
return "".join(
f"event: {event}\ndata: {json.dumps(payload)}\n\n"
for event, payload in _ANTHROPIC_SSE_EVENTS
).encode()
def _anthropic_sse_response(url: str) -> httpx.Response:
return httpx.Response(
status_code=200,
content=_anthropic_sse_bytes(),
headers={"content-type": "text/event-stream"},
request=httpx.Request("POST", url),
)
def test_mantle_completion_streaming_sends_stream_and_decodes_sse():
import litellm
requests = []
def mock_post(self, url, data=None, headers=None, **kwargs):
requests.append(_capture_request(url=url, headers=headers or {}, data=data))
return _anthropic_sse_response(url)
with patch("litellm.llms.custom_httpx.http_handler.HTTPHandler.post", mock_post):
response = litellm.completion(
model="bedrock/mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "ping"}],
max_tokens=10,
stream=True,
aws_access_key_id="fake-key",
aws_secret_access_key="fake-secret",
aws_region_name="us-east-1",
)
chunks = list(response)
assert len(requests) == 1
assert requests[0]["body"]["stream"] is True
content = "".join(chunk.choices[0].delta.content or "" for chunk in chunks)
assert content == "pong"
assert chunks[-1].choices[0].finish_reason == "stop"
@pytest.mark.asyncio
async def test_mantle_acompletion_streaming_sends_stream_and_decodes_sse():
import litellm
requests = []
async def mock_post(self, url, data=None, headers=None, **kwargs):
requests.append(_capture_request(url=url, headers=headers or {}, data=data))
return _anthropic_sse_response(url)
try:
with patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new=mock_post,
):
response = await litellm.acompletion(
model="bedrock/mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "ping"}],
max_tokens=10,
stream=True,
aws_access_key_id="fake-key",
aws_secret_access_key="fake-secret",
aws_region_name="us-east-1",
)
chunks = [chunk async for chunk in response]
finally:
await litellm.close_litellm_async_clients()
assert len(requests) == 1
assert requests[0]["body"]["stream"] is True
content = "".join(chunk.choices[0].delta.content or "" for chunk in chunks)
assert content == "pong"
assert chunks[-1].choices[0].finish_reason == "stop"
@pytest.mark.asyncio
async def test_mantle_anthropic_messages_streaming_sends_stream_and_passes_through_sse():
import litellm
requests = []
async def mock_post(self, url, data=None, headers=None, **kwargs):
requests.append(_capture_request(url=url, headers=headers or {}, data=data))
return _anthropic_sse_response(str(url))
try:
with patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new=mock_post,
):
response = await litellm.anthropic_messages(
model="bedrock/mantle/anthropic.claude-mythos-preview",
messages=[{"role": "user", "content": "ping"}],
max_tokens=10,
stream=True,
aws_access_key_id="fake-key",
aws_secret_access_key="fake-secret",
aws_region_name="us-east-1",
)
raw = b"".join([chunk async for chunk in response])
finally:
await litellm.close_litellm_async_clients()
assert len(requests) == 1
assert requests[0]["body"]["stream"] is True
text = raw.decode()
assert "event: message_start" in text
assert '"text": "pong"' in text
assert "event: message_stop" in text
Reference in a new issue View git blame Copy permalink