mirror of
https://github.com/tiennm99/litellm.git
synced 2026-07-16 12:16:41 +00:00
Move test to test_litellm/ folder
This commit is contained in:
Sameer Kankute
@@ -667,249 +667,4 @@ def test_get_supported_openai_params():
|
||||
assert "temperature" in params
|
||||
assert "stream" in params
|
||||
assert "background" in params
|
||||
assert "stream" in params
|
||||
|
||||
|
||||
class TestOpenAIFieldExclusionRegistry:
|
||||
"""Test suite for the OpenAI Field Exclusion Registry system"""
|
||||
|
||||
def setup_method(self):
|
||||
"""Setup test fixtures"""
|
||||
from litellm.llms.openai.responses.transformation import (
|
||||
OpenAIFieldExclusionRegistry,
|
||||
OpenAIResponsesAPIConfig
|
||||
)
|
||||
self.registry = OpenAIFieldExclusionRegistry
|
||||
self.config = OpenAIResponsesAPIConfig()
|
||||
|
||||
def test_registry_initialization(self):
|
||||
"""Test that the registry is properly initialized with ResponseReasoningItem"""
|
||||
# Test that we can get excluded fields (should not be empty if ResponseReasoningItem is registered)
|
||||
all_excluded_fields = self.registry.get_all_excluded_fields()
|
||||
|
||||
# The registry should have at least some fields if ResponseReasoningItem was successfully registered
|
||||
# If OpenAI SDK is not available, this might be empty, which is also valid
|
||||
assert isinstance(all_excluded_fields, set), "get_all_excluded_fields should return a set"
|
||||
|
||||
# If we have the OpenAI SDK available, we should have the expected fields
|
||||
try:
|
||||
from openai.types.responses import ResponseReasoningItem
|
||||
reasoning_fields = self.registry.get_excluded_fields_for_model(ResponseReasoningItem)
|
||||
expected_fields = {'status', 'content', 'encrypted_content'}
|
||||
assert expected_fields.issubset(reasoning_fields), f"Expected fields {expected_fields} to be subset of {reasoning_fields}"
|
||||
except ImportError:
|
||||
# If OpenAI SDK is not available, that's fine - the registry should handle this gracefully
|
||||
pytest.skip("OpenAI SDK not available, skipping ResponseReasoningItem specific tests")
|
||||
|
||||
def test_register_model_functionality(self):
|
||||
"""Test that we can register new models to the registry"""
|
||||
from pydantic import BaseModel
|
||||
from typing import Optional
|
||||
|
||||
# Create a test model with default None fields
|
||||
class TestResponseModel(BaseModel):
|
||||
id: str
|
||||
type: str = "test"
|
||||
status: Optional[str] = None
|
||||
content: Optional[str] = None
|
||||
required_field: str
|
||||
|
||||
# Register the test model
|
||||
self.registry.register_model(TestResponseModel)
|
||||
|
||||
# Verify it was registered and fields are detected
|
||||
excluded_fields = self.registry.get_excluded_fields_for_model(TestResponseModel)
|
||||
expected_excluded = {'status', 'content'} # Fields with default None
|
||||
|
||||
assert expected_excluded.issubset(excluded_fields), f"Expected {expected_excluded} to be in {excluded_fields}"
|
||||
assert 'id' not in excluded_fields, "Required field 'id' should not be excluded"
|
||||
assert 'required_field' not in excluded_fields, "Required field 'required_field' should not be excluded"
|
||||
|
||||
def test_get_all_excluded_fields(self):
|
||||
"""Test that get_all_excluded_fields aggregates fields from all registered models"""
|
||||
all_fields_before = self.registry.get_all_excluded_fields()
|
||||
|
||||
# Create and register a test model
|
||||
from pydantic import BaseModel
|
||||
from typing import Optional
|
||||
|
||||
class AnotherTestModel(BaseModel):
|
||||
id: str
|
||||
unique_field: Optional[str] = None
|
||||
|
||||
self.registry.register_model(AnotherTestModel)
|
||||
|
||||
all_fields_after = self.registry.get_all_excluded_fields()
|
||||
|
||||
# The new fields should be included
|
||||
assert 'unique_field' in all_fields_after, "New model's excluded field should be included"
|
||||
assert len(all_fields_after) >= len(all_fields_before), "Should have at least as many fields as before"
|
||||
|
||||
def test_convenience_registration_method(self):
|
||||
"""Test the convenience method for registering models"""
|
||||
from pydantic import BaseModel
|
||||
from typing import Optional
|
||||
|
||||
class ConvenienceTestModel(BaseModel):
|
||||
id: str
|
||||
convenience_field: Optional[str] = None
|
||||
|
||||
# Use the convenience method
|
||||
self.config.register_model_for_field_exclusion(ConvenienceTestModel)
|
||||
|
||||
# Verify it was registered
|
||||
excluded_fields = self.registry.get_excluded_fields_for_model(ConvenienceTestModel)
|
||||
assert 'convenience_field' in excluded_fields, "Field should be excluded after registration"
|
||||
|
||||
def test_field_filtering_with_registry(self):
|
||||
"""Test that the field filtering works correctly with the registry"""
|
||||
|
||||
# Test data that matches the structure of ResponseReasoningItem
|
||||
test_input = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": "test message"
|
||||
},
|
||||
{
|
||||
"id": "reasoning-123",
|
||||
"type": "reasoning",
|
||||
"status": None, # Should be filtered out
|
||||
"content": None, # Should be filtered out
|
||||
"encrypted_content": None, # Should be filtered out
|
||||
"summary": [{"text": "This reasoning shows...", "type": "summary_text"}],
|
||||
"role": "assistant"
|
||||
},
|
||||
{
|
||||
"id": "message-456",
|
||||
"type": "message",
|
||||
"status": "completed", # Should be preserved (not None)
|
||||
"content": "Hello! How can I help?", # Should be preserved (not None)
|
||||
"role": "assistant"
|
||||
}
|
||||
]
|
||||
|
||||
# Process the input through the validation
|
||||
result = self.config._validate_input_param(test_input)
|
||||
|
||||
# Verify the structure
|
||||
assert len(result) == 3, "Should have 3 items"
|
||||
|
||||
# Check the reasoning item (index 1)
|
||||
reasoning_item = result[1]
|
||||
assert reasoning_item["type"] == "reasoning"
|
||||
assert reasoning_item["id"] == "reasoning-123"
|
||||
assert "summary" in reasoning_item, "summary field should be preserved"
|
||||
assert "role" in reasoning_item, "role field should be preserved"
|
||||
|
||||
# These fields should be filtered out if they are in the registry
|
||||
all_excluded_fields = self.registry.get_all_excluded_fields()
|
||||
if 'status' in all_excluded_fields:
|
||||
assert "status" not in reasoning_item, "status field should be filtered out"
|
||||
if 'content' in all_excluded_fields:
|
||||
assert "content" not in reasoning_item, "content field should be filtered out"
|
||||
if 'encrypted_content' in all_excluded_fields:
|
||||
assert "encrypted_content" not in reasoning_item, "encrypted_content field should be filtered out"
|
||||
|
||||
# Check the message item (index 2) - non-None values should be preserved
|
||||
message_item = result[2]
|
||||
assert message_item["type"] == "message"
|
||||
assert message_item["status"] == "completed", "Non-None status should be preserved"
|
||||
assert message_item["content"] == "Hello! How can I help?", "Non-None content should be preserved"
|
||||
|
||||
def test_field_filtering_with_empty_registry(self):
|
||||
"""Test that filtering works gracefully when no models are registered"""
|
||||
# Create a fresh registry for this test
|
||||
from litellm.llms.openai.responses.transformation import OpenAIFieldExclusionRegistry
|
||||
|
||||
# Save the current state
|
||||
original_models = OpenAIFieldExclusionRegistry._MODELS_REQUIRING_EXCLUSION.copy()
|
||||
|
||||
try:
|
||||
# Clear the registry
|
||||
OpenAIFieldExclusionRegistry._MODELS_REQUIRING_EXCLUSION.clear()
|
||||
|
||||
# Test data
|
||||
test_input = [{
|
||||
"id": "test-123",
|
||||
"status": None,
|
||||
"content": None,
|
||||
"other_field": "should be preserved"
|
||||
}]
|
||||
|
||||
# Process the input
|
||||
result = self.config._validate_input_param(test_input)
|
||||
|
||||
# With empty registry, nothing should be filtered (all fields preserved)
|
||||
assert len(result) == 1
|
||||
item = result[0]
|
||||
assert "status" in item, "With empty registry, status should be preserved"
|
||||
assert "content" in item, "With empty registry, content should be preserved"
|
||||
assert item["other_field"] == "should be preserved"
|
||||
|
||||
finally:
|
||||
# Restore the original state
|
||||
OpenAIFieldExclusionRegistry._MODELS_REQUIRING_EXCLUSION = original_models
|
||||
|
||||
def test_pydantic_v1_v2_compatibility(self):
|
||||
"""Test that the registry works with both Pydantic v1 and v2"""
|
||||
from pydantic import BaseModel
|
||||
from typing import Optional
|
||||
|
||||
class CompatibilityTestModel(BaseModel):
|
||||
id: str
|
||||
optional_field: Optional[str] = None
|
||||
required_field: str = "default"
|
||||
|
||||
# Register the model
|
||||
self.registry.register_model(CompatibilityTestModel)
|
||||
|
||||
# Get excluded fields
|
||||
excluded_fields = self.registry.get_excluded_fields_for_model(CompatibilityTestModel)
|
||||
|
||||
# Should work regardless of Pydantic version
|
||||
assert isinstance(excluded_fields, set), "Should return a set"
|
||||
assert 'optional_field' in excluded_fields, "Field with default None should be excluded"
|
||||
|
||||
# Test that the model fields are accessible (works in both v1 and v2)
|
||||
model_fields = getattr(CompatibilityTestModel, "model_fields", None)
|
||||
if model_fields is None:
|
||||
model_fields = getattr(CompatibilityTestModel, "__fields__", {})
|
||||
assert len(model_fields) > 0, "Should be able to access model fields"
|
||||
|
||||
def test_non_registered_model_returns_empty_set(self):
|
||||
"""Test that non-registered models return empty excluded fields"""
|
||||
from pydantic import BaseModel
|
||||
|
||||
class UnregisteredModel(BaseModel):
|
||||
id: str
|
||||
some_field: str = None
|
||||
|
||||
# Don't register this model
|
||||
excluded_fields = self.registry.get_excluded_fields_for_model(UnregisteredModel)
|
||||
|
||||
assert excluded_fields == set(), "Non-registered model should return empty set"
|
||||
|
||||
@pytest.mark.parametrize("field_value", [None, "", 0, False, []])
|
||||
def test_only_none_values_are_filtered(self, field_value):
|
||||
"""Test that only None values are filtered, not other falsy values"""
|
||||
test_input = [{
|
||||
"id": "test-123",
|
||||
"status": field_value,
|
||||
"content": "actual content",
|
||||
"other_field": "preserved"
|
||||
}]
|
||||
|
||||
result = self.config._validate_input_param(test_input)
|
||||
item = result[0]
|
||||
|
||||
if field_value is None:
|
||||
# Only None should be filtered (if status is in the registry)
|
||||
all_excluded_fields = self.registry.get_all_excluded_fields()
|
||||
if 'status' in all_excluded_fields:
|
||||
assert "status" not in item, f"None value should be filtered out"
|
||||
else:
|
||||
assert item["status"] is None, f"If not in registry, None should be preserved"
|
||||
else:
|
||||
# Other falsy values should be preserved
|
||||
assert "status" in item, f"Non-None value {field_value} should be preserved"
|
||||
assert item["status"] == field_value, f"Value should be exactly {field_value}"
|
||||
assert "stream" in params
|
||||
Reference in New Issue
Block a user