mirror of
https://github.com/tiennm99/litellm.git
synced 2026-08-17 08:25:03 +00:00
feat: Add OVHCloud AI Endpoints as a provider
This commit is contained in:
Elias TOURNEUX
@@ -0,0 +1,256 @@
|
||||
"""
|
||||
Unit tests for OVHCloud AI Endpoints chat integration.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.llms.ovhcloud.utils import OVHCloudException
|
||||
|
||||
sys.path.insert(
|
||||
0, os.path.abspath("../../../../..")
|
||||
) # Adds the parent directory to the system path
|
||||
|
||||
from litellm.llms.ovhcloud.chat.transformation import (
|
||||
OVHCloudChatCompletionStreamingHandler,
|
||||
OVHCloudChatConfig,
|
||||
)
|
||||
|
||||
config = OVHCloudChatConfig()
|
||||
model = "ovhcloud/Mistral-7B-Instruct-v0.3"
|
||||
|
||||
class TestOvhCloudChatCompletionStreamingHandler:
|
||||
def test_chunk_parser_successful(self):
|
||||
handler = OVHCloudChatCompletionStreamingHandler(
|
||||
streaming_response=None, sync_stream=True
|
||||
)
|
||||
|
||||
chunk = {
|
||||
"id": "test_id",
|
||||
"created": 1234567890,
|
||||
"model": "gpt-oss-20b",
|
||||
"usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30},
|
||||
"choices": [
|
||||
{"delta": {"content": "test content", "reasoning": "test reasoning"}}
|
||||
],
|
||||
}
|
||||
|
||||
result = handler.chunk_parser(chunk)
|
||||
|
||||
assert result.id == "test_id"
|
||||
assert result.object == "chat.completion.chunk"
|
||||
assert result.created == 1234567890
|
||||
assert result.model == "gpt-oss-20b"
|
||||
assert result.usage.prompt_tokens == chunk["usage"]["prompt_tokens"]
|
||||
assert result.usage.completion_tokens == chunk["usage"]["completion_tokens"]
|
||||
assert result.usage.total_tokens == chunk["usage"]["total_tokens"]
|
||||
assert len(result.choices) == 1
|
||||
assert result.choices[0]["delta"]["reasoning_content"] == "test reasoning"
|
||||
|
||||
def test_chunk_parser_error_response(self):
|
||||
handler = OVHCloudChatCompletionStreamingHandler(
|
||||
streaming_response=None, sync_stream=True
|
||||
)
|
||||
|
||||
error_chunk = {
|
||||
"error": {
|
||||
"message": "test error",
|
||||
"code": 400,
|
||||
}
|
||||
}
|
||||
|
||||
with pytest.raises(OVHCloudException) as exc_info:
|
||||
handler.chunk_parser(error_chunk)
|
||||
|
||||
assert "OVHCloud Error: test error" in str(exc_info.value)
|
||||
assert exc_info.value.status_code == 400
|
||||
|
||||
def test_chunk_parser_key_error(self):
|
||||
handler = OVHCloudChatCompletionStreamingHandler(
|
||||
streaming_response=None, sync_stream=True
|
||||
)
|
||||
|
||||
invalid_chunk = {"incomplete": "data"}
|
||||
|
||||
with pytest.raises(OVHCloudException) as exc_info:
|
||||
handler.chunk_parser(invalid_chunk)
|
||||
|
||||
assert "KeyError" in str(exc_info.value)
|
||||
assert exc_info.value.status_code == 400
|
||||
|
||||
|
||||
class TestOVHCloudConfig:
|
||||
def test_transform_request_basic(self):
|
||||
"""Test basic request transformation"""
|
||||
transformed_request = config.transform_request(
|
||||
model,
|
||||
messages=[
|
||||
{"role": "user", "content": "Hello, world!"}
|
||||
],
|
||||
optional_params={},
|
||||
litellm_params={},
|
||||
headers={},
|
||||
)
|
||||
|
||||
assert transformed_request["model"] == model
|
||||
assert transformed_request["messages"] == [
|
||||
{"role": "user", "content": "Hello, world!"}
|
||||
]
|
||||
|
||||
def test_transform_request_with_extra_body(self):
|
||||
"""Test request transformation with extra_body parameters"""
|
||||
transformed_request = config.transform_request(
|
||||
model,
|
||||
messages=[{"role": "user", "content": "Hello, world!"}],
|
||||
optional_params={"extra_body": {"custom_param": "custom_value"}},
|
||||
litellm_params={},
|
||||
headers={},
|
||||
)
|
||||
|
||||
assert transformed_request["custom_param"] == "custom_value"
|
||||
assert transformed_request["messages"] == [
|
||||
{"role": "user", "content": "Hello, world!"}
|
||||
]
|
||||
|
||||
def test_map_openai_params(self):
|
||||
"""Test OpenAI parameter mapping"""
|
||||
non_default_params = {
|
||||
"temperature": 0.7,
|
||||
"max_tokens": 100,
|
||||
"top_p": 0.9,
|
||||
}
|
||||
|
||||
mapped_params = config.map_openai_params(
|
||||
non_default_params=non_default_params,
|
||||
optional_params={},
|
||||
model=model,
|
||||
drop_params=False,
|
||||
)
|
||||
|
||||
assert mapped_params["temperature"] == 0.7
|
||||
assert mapped_params["max_tokens"] == 100
|
||||
assert mapped_params["top_p"] == 0.9
|
||||
|
||||
def test_get_error_class(self):
|
||||
"""Test error class creation"""
|
||||
error = config.get_error_class(
|
||||
error_message="Test error",
|
||||
status_code=400,
|
||||
headers={"Content-Type": "application/json"}
|
||||
)
|
||||
|
||||
assert isinstance(error, OVHCloudException)
|
||||
assert error.message == "Test error"
|
||||
assert error.status_code == 400
|
||||
|
||||
|
||||
def test_ovhcloud_integration():
|
||||
import os
|
||||
from litellm import completion
|
||||
|
||||
api_key = os.getenv("OVHCLOUD_API_KEY")
|
||||
|
||||
if not api_key:
|
||||
pytest.skip("OVHCLOUD_API_KEY not set, skipping test")
|
||||
|
||||
response = completion(
|
||||
model,
|
||||
messages=[{"role": "user", "content": "Say hello in one word"}],
|
||||
api_key=api_key,
|
||||
max_tokens=10,
|
||||
temperature=0.7
|
||||
)
|
||||
|
||||
assert response.choices[0].message.content
|
||||
assert len(response.choices[0].message.content.strip()) > 0
|
||||
assert response.model
|
||||
assert response.usage
|
||||
assert response.usage.total_tokens > 0
|
||||
|
||||
def test_OVHCloud_streaming_integration():
|
||||
"""
|
||||
Integration test for streaming - requires real API key
|
||||
Run with: pytest -k test_OVHCloud_streaming_integration -s
|
||||
"""
|
||||
import os
|
||||
from litellm import completion
|
||||
|
||||
api_key = os.getenv("OVHCLOUD_API_KEY")
|
||||
|
||||
if not api_key:
|
||||
pytest.skip("OVHCLOUD_API_KEY not set, skipping test")
|
||||
|
||||
try:
|
||||
print(f"🔍 Testing streaming with API key: {api_key[:6]}...{api_key[-4:]} (length: {len(api_key)})")
|
||||
print(f"🔍 API base URL: {os.getenv('OVHCLOUD_API_BASE')}")
|
||||
|
||||
response = completion(
|
||||
model,
|
||||
messages=[{"role": "user", "content": "Count from 1 to 5"}],
|
||||
api_key=api_key,
|
||||
max_tokens=50,
|
||||
stream=True
|
||||
)
|
||||
|
||||
chunks = []
|
||||
content_parts = []
|
||||
|
||||
for chunk in response:
|
||||
chunks.append(chunk)
|
||||
if chunk.choices[0].delta.content:
|
||||
content_parts.append(chunk.choices[0].delta.content)
|
||||
|
||||
assert len(chunks) > 0, "Should receive at least one chunk"
|
||||
assert len(content_parts) > 0, "Should receive content in chunks"
|
||||
|
||||
full_content = "".join(content_parts)
|
||||
assert len(full_content.strip()) > 0, "Should have non-empty content"
|
||||
|
||||
print(f"✅ Received {len(chunks)} chunks")
|
||||
print(f"✅ Full content: {full_content}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Streaming integration test error details:")
|
||||
print(f" Error type: {type(e).__name__}")
|
||||
print(f" Error message: {str(e)}")
|
||||
if hasattr(e, 'status_code'):
|
||||
print(f" Status code: {e.status_code}")
|
||||
if hasattr(e, 'response'):
|
||||
print(f" Response: {e.response}")
|
||||
|
||||
pytest.fail(f"Streaming integration test failed: {type(e).__name__}: {str(e)}")
|
||||
|
||||
def test_ovhcloud_with_custom_base_url():
|
||||
"""
|
||||
Test OVHCloud with custom base URL
|
||||
"""
|
||||
import os
|
||||
from litellm import completion
|
||||
|
||||
api_key = os.getenv("OVHCLOUD_API_KEY")
|
||||
|
||||
if not api_key:
|
||||
pytest.skip("OVHCLOUD_API_KEY not set, skipping test")
|
||||
|
||||
custom_base_url = os.getenv("OVHCLOUD_API_BASE", "https://oai.endpoints.kepler.ai.cloud.ovh.net/v1")
|
||||
|
||||
try:
|
||||
response = completion(
|
||||
model,
|
||||
messages=[{"role": "user", "content": "Hello"}],
|
||||
api_key=api_key,
|
||||
api_base=custom_base_url,
|
||||
max_tokens=5
|
||||
)
|
||||
|
||||
assert response.choices[0].message.content
|
||||
print(f"✅ Custom base URL test passed: {response.choices[0].message.content}")
|
||||
|
||||
except Exception as e:
|
||||
pytest.fail(f"Custom base URL test failed: {str(e)}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
@@ -0,0 +1,37 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
import litellm
|
||||
|
||||
model="ovhcloud/BGE-M3"
|
||||
|
||||
def mock_embedding_response(*args, **kwargs):
|
||||
class MockResponse:
|
||||
def __init__(self):
|
||||
self.data = [{"embedding": [0.1, 0.2, 0.3]}]
|
||||
self.usage = litellm.Usage()
|
||||
self.model = kwargs.get("model", model)
|
||||
self.object = "embedding"
|
||||
|
||||
def __getitem__(self, key):
|
||||
return getattr(self, key)
|
||||
|
||||
return MockResponse()
|
||||
|
||||
|
||||
def test_ovhcloud_embeddings():
|
||||
with patch("litellm.embedding", side_effect=mock_embedding_response) as mock_embed:
|
||||
response = litellm.embedding(
|
||||
model,
|
||||
input=["good morning from litellm"],
|
||||
)
|
||||
|
||||
mock_embed.assert_called_once_with(
|
||||
model,
|
||||
input=["good morning from litellm"],
|
||||
)
|
||||
|
||||
assert isinstance(response.data, list)
|
||||
assert "embedding" in response.data[0]
|
||||
assert isinstance(response.data[0]["embedding"], list)
|
||||
assert response.model == model
|
||||
assert response.object == "embedding"
|
||||
Reference in New Issue
Block a user