scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-08 22:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 78

fix(minimax): harden video adapter checks

Browse source
This commit is contained in:
octo-patch 2026-08-06 16:52:09 +08:00
parent 74bab168c9
commit 915348690e
No known key found for this signature in database
2 changed files with 567 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

135
litellm/llms/minimax/videos/transformation.py
View file

@ -1,12 +1,16 @@
"""MiniMax v1 video generation transformations."""
from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
import base64
from os import PathLike
from typing import TYPE_CHECKING
from urllib.parse import quote, urlsplit, urlunsplit
import httpx
from httpx._types import RequestFiles
import litellm
from litellm.images.utils import ImageEditRequestUtils
from litellm.litellm_core_utils.url_utils import async_safe_get, safe_get
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.videos.transformation import BaseVideoConfig
from litellm.llms.custom_httpx.http_handler import (
@ -28,7 +32,7 @@ if TYPE_CHECKING:
LiteLLMLoggingObj = _LiteLLMLoggingObj
else:
LiteLLMLoggingObj = Any
LiteLLMLoggingObj = object
class MinimaxVideoConfig(BaseVideoConfig):
@ -56,14 +60,14 @@ class MinimaxVideoConfig(BaseVideoConfig):
video_create_optional_params: VideoCreateOptionalRequestParams,
model: str,
drop_params: bool,
) -> Dict:
mapped_params: Dict[str, Any] = {}
) -> dict:
mapped_params: dict = {}
for key, value in video_create_optional_params.items():
if value is None or key in {"model", "prompt", "extra_headers", "user"}:
continue
if key == "input_reference":
mapped_params["first_frame_image"] = value
mapped_params["first_frame_image"] = self._prepare_first_frame_image(value)
elif key == "seconds":
mapped_params["duration"] = self._coerce_duration(value)
elif key == "size":
@ -81,8 +85,8 @@ class MinimaxVideoConfig(BaseVideoConfig):
self,
headers: dict,
model: str,
api_key: Optional[str] = None,
litellm_params: Optional[GenericLiteLLMParams] = None,
api_key: str | None = None,
litellm_params: GenericLiteLLMParams | None = None,
) -> dict:
if litellm_params and litellm_params.api_key:
api_key = api_key or litellm_params.api_key
@ -104,13 +108,13 @@ class MinimaxVideoConfig(BaseVideoConfig):
def get_complete_url(
self,
model: str,
api_base: Optional[str],
api_base: str | None,
litellm_params: dict,
) -> str:
"""Return the regional MiniMax v1 root used by all video operations."""
base_url = api_base or get_secret_str("MINIMAX_API_BASE") or "https://api.minimax.io/v1"
base_url = base_url.rstrip("/")
for suffix in ("/video_generation", "/query/video_generation", "/files/retrieve"):
for suffix in ("/query/video_generation", "/video_generation", "/files/retrieve"):
if base_url.endswith(suffix):
base_url = base_url[: -len(suffix)]
break
@ -123,11 +127,11 @@ class MinimaxVideoConfig(BaseVideoConfig):
model: str,
prompt: str,
api_base: str,
video_create_optional_request_params: Dict,
video_create_optional_request_params: dict,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Tuple[Dict, RequestFiles, str]:
request_data: Dict[str, Any] = {"model": model, "prompt": prompt}
) -> tuple[dict, RequestFiles, str]:
request_data: dict = {"model": model, "prompt": prompt}
request_data.update(video_create_optional_request_params)
request_data.pop("extra_headers", None)
request_data.pop("extra_body", None)
@ -139,8 +143,8 @@ class MinimaxVideoConfig(BaseVideoConfig):
model: str,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
custom_llm_provider: Optional[str] = None,
request_data: Optional[Dict] = None,
custom_llm_provider: str | None = None,
request_data: dict | None = None,
) -> VideoObject:
response_data = self._parse_json_response(raw_response)
self._raise_for_provider_error(raw_response, response_data)
@ -148,7 +152,7 @@ class MinimaxVideoConfig(BaseVideoConfig):
if task_id is None:
raise ValueError("MiniMax did not return a task_id for video generation")
video_data: Dict[str, Any] = {
video_data: dict = {
"id": str(task_id),
"object": "video",
"status": self._map_status(response_data.get("status", "queueing")),
@ -166,8 +170,8 @@ class MinimaxVideoConfig(BaseVideoConfig):
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
variant: Optional[str] = None,
) -> Tuple[str, Dict]:
variant: str | None = None,
) -> tuple[str, dict]:
task_id = quote(extract_original_video_id(video_id), safe="")
return f"{api_base.rstrip('/')}/query/video_generation?task_id={task_id}", {}
@ -198,7 +202,7 @@ class MinimaxVideoConfig(BaseVideoConfig):
if not download_url:
raise ValueError("MiniMax did not return a video download URL")
video_response = client.get(download_url)
video_response = safe_get(client, download_url)
self._raise_for_status(video_response)
return video_response.content
@ -229,7 +233,7 @@ class MinimaxVideoConfig(BaseVideoConfig):
if not download_url:
raise ValueError("MiniMax did not return a video download URL")
video_response = await client.get(download_url)
video_response = await async_safe_get(client, download_url)
self._raise_for_status(video_response)
return video_response.content
@ -239,7 +243,7 @@ class MinimaxVideoConfig(BaseVideoConfig):
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Tuple[str, Dict]:
) -> tuple[str, dict]:
task_id = quote(extract_original_video_id(video_id), safe="")
return f"{api_base.rstrip('/')}/query/video_generation?task_id={task_id}", {}
@ -247,7 +251,7 @@ class MinimaxVideoConfig(BaseVideoConfig):
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
custom_llm_provider: Optional[str] = None,
custom_llm_provider: str | None = None,
) -> VideoObject:
response_data = self._parse_json_response(raw_response)
self._raise_for_provider_error(raw_response, response_data)
@ -255,7 +259,7 @@ class MinimaxVideoConfig(BaseVideoConfig):
if task_id is None:
raise ValueError("MiniMax did not return a task_id for video status")
model = response_data.get("model")
video_data: Dict[str, Any] = {
video_data: dict = {
"id": str(task_id),
"object": "video",
"status": self._map_status(response_data.get("status", "processing")),
@ -282,15 +286,15 @@ class MinimaxVideoConfig(BaseVideoConfig):
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
extra_body: Optional[Dict[str, Any]] = None,
) -> Tuple[str, Dict]:
extra_body: dict | None = None,
) -> tuple[str, dict]:
raise NotImplementedError("Video remix is not supported by the MiniMax v1 API")
def transform_video_remix_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
custom_llm_provider: Optional[str] = None,
custom_llm_provider: str | None = None,
) -> VideoObject:
raise NotImplementedError("Video remix is not supported by the MiniMax v1 API")
@ -299,19 +303,19 @@ class MinimaxVideoConfig(BaseVideoConfig):
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
after: Optional[str] = None,
limit: Optional[int] = None,
order: Optional[str] = None,
extra_query: Optional[Dict[str, Any]] = None,
) -> Tuple[str, Dict]:
after: str | None = None,
limit: int | None = None,
order: str | None = None,
extra_query: dict | None = None,
) -> tuple[str, dict]:
raise NotImplementedError("Video listing is not supported by the MiniMax v1 API")
def transform_video_list_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLoggingObj,
custom_llm_provider: Optional[str] = None,
) -> Dict[str, str]:
custom_llm_provider: str | None = None,
) -> dict[str, str]:
raise NotImplementedError("Video listing is not supported by the MiniMax v1 API")
def transform_video_delete_request(
@ -320,7 +324,7 @@ class MinimaxVideoConfig(BaseVideoConfig):
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> Tuple[str, Dict]:
) -> tuple[str, dict]:
raise NotImplementedError("Video deletion is not supported by the MiniMax v1 API")
def transform_video_delete_response(
@ -330,20 +334,18 @@ class MinimaxVideoConfig(BaseVideoConfig):
) -> VideoObject:
raise NotImplementedError("Video deletion is not supported by the MiniMax v1 API")
def get_error_class(
self, error_message: str, status_code: int, headers: Union[dict, httpx.Headers]
) -> BaseLLMException:
def get_error_class(self, error_message: str, status_code: int, headers: dict | httpx.Headers) -> BaseLLMException:
return BaseLLMException(status_code=status_code, message=error_message, headers=headers)
@staticmethod
def _coerce_duration(value: Any) -> Any:
def _coerce_duration(value: object) -> object:
try:
return int(float(value))
except (TypeError, ValueError):
return value
@staticmethod
def _map_status(status: Any) -> str:
def _map_status(status: object) -> str:
normalized = str(status or "").strip().lower().replace(" ", "_")
if normalized in {"success", "succeeded", "completed", "complete"}:
return "completed"
@ -354,7 +356,7 @@ class MinimaxVideoConfig(BaseVideoConfig):
return "in_progress"
@staticmethod
def _add_request_metadata(video_data: Dict[str, Any], request_data: Optional[Dict]) -> None:
def _add_request_metadata(video_data: dict, request_data: dict | None) -> None:
if not request_data:
return
if request_data.get("duration") is not None:
@ -363,7 +365,7 @@ class MinimaxVideoConfig(BaseVideoConfig):
video_data["size"] = str(request_data["resolution"])
@staticmethod
def _usage_from_video(video_obj: VideoObject) -> Dict[str, Any]:
def _usage_from_video(video_obj: VideoObject) -> dict:
if video_obj.seconds is None:
return {}
try:
@ -372,18 +374,18 @@ class MinimaxVideoConfig(BaseVideoConfig):
return {}
@staticmethod
def _wrap_video_id(video_obj: VideoObject, provider: Optional[str], model: Optional[str]) -> None:
def _wrap_video_id(video_obj: VideoObject, provider: str | None, model: str | None) -> None:
if provider and video_obj.id:
video_obj.id = encode_video_id_with_provider(video_obj.id, provider, model)
@staticmethod
def _parse_json_response(raw_response: httpx.Response) -> Dict[str, Any]:
def _parse_json_response(self, raw_response: httpx.Response) -> dict:
self._raise_for_status(raw_response)
try:
return raw_response.json()
except Exception as exc:
raise ValueError(f"MiniMax returned an invalid JSON response: {exc}") from exc
def _raise_for_provider_error(self, raw_response: httpx.Response, response_data: Dict[str, Any]) -> None:
def _raise_for_provider_error(self, raw_response: httpx.Response, response_data: dict) -> None:
self._raise_for_status(raw_response)
base_resp = response_data.get("base_resp") or {}
status_code = base_resp.get("status_code")
@ -396,8 +398,8 @@ class MinimaxVideoConfig(BaseVideoConfig):
raise self.get_error_class(raw_response.text, raw_response.status_code, raw_response.headers)
@staticmethod
def _request_headers(raw_response: httpx.Response) -> Dict[str, str]:
request = getattr(raw_response, "request", None)
def _request_headers(raw_response: httpx.Response) -> dict[str, str]:
request = getattr(raw_response, "_request", None)
request_headers = getattr(request, "headers", None)
if isinstance(request_headers, (dict, httpx.Headers)):
authorization = request_headers.get("Authorization")
@ -407,8 +409,8 @@ class MinimaxVideoConfig(BaseVideoConfig):
@staticmethod
def _api_base_from_response(raw_response: httpx.Response) -> str:
request = getattr(raw_response, "request", None)
request_url = getattr(request, "url", None) or getattr(raw_response, "url", None)
request = getattr(raw_response, "_request", None)
request_url = getattr(request, "url", None)
if request_url is None:
return "https://api.minimax.io/v1"
parsed = urlsplit(str(request_url))
@ -423,7 +425,7 @@ class MinimaxVideoConfig(BaseVideoConfig):
return content_type.startswith("video/") or content_type == "application/octet-stream"
@staticmethod
def _get_download_url(response_data: Dict[str, Any]) -> Optional[str]:
def _get_download_url(response_data: dict) -> str | None:
file_data = response_data.get("file")
if isinstance(file_data, dict):
for key in ("download_url", "url"):
@ -433,3 +435,38 @@ class MinimaxVideoConfig(BaseVideoConfig):
if response_data.get(key):
return str(response_data[key])
return None
@staticmethod
def _prepare_first_frame_image(image: object) -> object:
if isinstance(image, str):
return image
content = image
content_type = None
if isinstance(image, tuple):
if len(image) < 2:
raise ValueError("MiniMax input_reference tuple must include file content")
content = image[1]
if len(image) >= 3 and isinstance(image[2], str):
content_type = image[2]
if isinstance(content, PathLike):
with open(content, "rb") as image_file:
image_bytes = image_file.read()
elif isinstance(content, bytes):
image_bytes = content
elif hasattr(content, "read"):
current_position = content.tell() if hasattr(content, "tell") else None
if hasattr(content, "seek"):
content.seek(0)
image_bytes = content.read()
if current_position is not None and hasattr(content, "seek"):
content.seek(current_position)
else:
raise TypeError("MiniMax input_reference must be a URL, path, bytes, or file object")
if not isinstance(image_bytes, bytes):
raise TypeError("MiniMax input_reference file content must be bytes")
content_type = content_type or ImageEditRequestUtils.get_image_content_type(image_bytes)
encoded = base64.b64encode(image_bytes).decode("ascii")
return f"data:{content_type};base64,{encoded}"

487
tests/test_litellm/llms/minimax/videos/test_minimax_video_transformation.py
View file

@ -1,8 +1,10 @@
"""Tests for MiniMax v1 video generation transformations."""
from unittest.mock import Mock, patch
from io import BytesIO
from unittest.mock import AsyncMock, Mock, patch
import httpx
import pytest
import litellm
from litellm.llms.minimax.videos.transformation import MinimaxVideoConfig
@ -24,6 +26,88 @@ class TestMinimaxVideoTransformation:
)
assert isinstance(config, MinimaxVideoConfig)
def test_supported_params_and_environment(self):
assert self.config.get_supported_openai_params("MiniMax-Hailuo-2.3") == [
"model",
"prompt",
"input_reference",
"seconds",
"size",
"user",
"extra_headers",
"extra_body",
"prompt_optimizer",
"fast_pretreatment",
"duration",
"resolution",
"callback_url",
]
headers = self.config.validate_environment(
headers={"X-Test": "value"},
model="MiniMax-Hailuo-2.3",
litellm_params=GenericLiteLLMParams(api_key="params-key"),
)
assert headers == {
"X-Test": "value",
"Authorization": "Bearer params-key",
"Content-Type": "application/json",
}
explicit_headers = self.config.validate_environment(
headers={},
model="MiniMax-Hailuo-2.3",
api_key="explicit-key",
litellm_params=GenericLiteLLMParams(api_key="params-key"),
)
assert explicit_headers["Authorization"] == "Bearer explicit-key"
with (
patch.object(litellm, "api_key", None),
patch(
"litellm.llms.minimax.videos.transformation.get_secret_str",
return_value=None,
),
pytest.raises(ValueError, match="MiniMax API key is required"),
):
self.config.validate_environment(
headers={},
model="MiniMax-Hailuo-2.3",
litellm_params=GenericLiteLLMParams(),
)
@pytest.mark.parametrize(
("api_base", "expected"),
[
("https://api.minimax.io/v1/video_generation", "https://api.minimax.io/v1"),
("https://api.minimaxi.com/v1/query/video_generation", "https://api.minimaxi.com/v1"),
("https://api.minimax.io/custom/", "https://api.minimax.io/custom/v1"),
],
)
def test_get_complete_url(self, api_base, expected):
assert (
self.config.get_complete_url(
model="MiniMax-Hailuo-2.3",
api_base=api_base,
litellm_params={},
)
== expected
)
def test_get_complete_url_uses_configured_default(self):
with patch(
"litellm.llms.minimax.videos.transformation.get_secret_str",
return_value="https://api.minimaxi.com/v1/files/retrieve",
):
assert (
self.config.get_complete_url(
model="MiniMax-Hailuo-2.3",
api_base=None,
litellm_params={},
)
== "https://api.minimaxi.com/v1"
)
def test_transform_create_request_maps_text_and_image_parameters(self):
params = self.config.map_openai_params(
{
@ -55,6 +139,121 @@ class TestMinimaxVideoTransformation:
assert files == []
assert url == "https://api.minimax.io/v1/video_generation"
def test_file_inputs_are_encoded_as_data_urls(self, tmp_path):
image_bytes = b"\x89PNG\r\n\x1a\nimage"
byte_params = self.config.map_openai_params(
{"input_reference": image_bytes},
model="MiniMax-Hailuo-2.3",
drop_params=False,
)
assert byte_params["first_frame_image"].startswith("data:image/png;base64,")
image_file = BytesIO(image_bytes)
image_file.seek(2)
file_params = self.config.map_openai_params(
{"input_reference": image_file},
model="MiniMax-Hailuo-2.3",
drop_params=False,
)
assert file_params["first_frame_image"] == byte_params["first_frame_image"]
assert image_file.tell() == 2
tuple_params = self.config.map_openai_params(
{"input_reference": ("frame.webp", image_bytes, "image/webp")},
model="MiniMax-Hailuo-2.3",
drop_params=False,
)
assert tuple_params["first_frame_image"].startswith("data:image/webp;base64,")
image_path = tmp_path / "frame.png"
image_path.write_bytes(image_bytes)
path_params = self.config.map_openai_params(
{"input_reference": image_path},
model="MiniMax-Hailuo-2.3",
drop_params=False,
)
assert path_params["first_frame_image"] == byte_params["first_frame_image"]
def test_invalid_file_inputs_are_rejected(self):
with pytest.raises(ValueError, match="tuple must include file content"):
self.config.map_openai_params(
{"input_reference": ("frame.png",)},
model="MiniMax-Hailuo-2.3",
drop_params=False,
)
with pytest.raises(TypeError, match="URL, path, bytes, or file object"):
self.config.map_openai_params(
{"input_reference": object()},
model="MiniMax-Hailuo-2.3",
drop_params=False,
)
text_file = Mock()
text_file.read.return_value = "not-bytes"
with pytest.raises(TypeError, match="file content must be bytes"):
self.config.map_openai_params(
{"input_reference": text_file},
model="MiniMax-Hailuo-2.3",
drop_params=False,
)
def test_map_params_skips_empty_values_and_preserves_provider_fields(self):
params = self.config.map_openai_params(
{
"model": "ignored",
"prompt": "ignored",
"user": "ignored",
"extra_headers": {"X-Test": "ignored"},
"seconds": "not-a-number",
"prompt_optimizer": False,
"callback_url": None,
"extra_body": {
"fast_pretreatment": True,
"prompt_optimizer": None,
},
},
model="MiniMax-Hailuo-2.3",
drop_params=False,
)
assert params == {
"duration": "not-a-number",
"prompt_optimizer": False,
"fast_pretreatment": True,
}
assert (
self.config.map_openai_params(
{"extra_body": "ignored"},
model="MiniMax-Hailuo-2.3",
drop_params=False,
)
== {}
)
def test_transform_create_request_removes_sdk_only_fields(self):
data, files, url = self.config.transform_video_create_request(
model="MiniMax-Hailuo-2.3",
prompt="A city at sunrise",
api_base="https://api.minimax.io/v1/",
video_create_optional_request_params={
"duration": 6,
"extra_headers": {"X-Test": "value"},
"extra_body": {"ignored": True},
"user": "user-123",
},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert data == {
"model": "MiniMax-Hailuo-2.3",
"prompt": "A city at sunrise",
"duration": 6,
}
assert files == []
assert url == "https://api.minimax.io/v1/video_generation"
def test_create_response_wraps_task_id_and_maps_status(self):
response = httpx.Response(
200,
@ -76,6 +275,44 @@ class TestMinimaxVideoTransformation:
assert result.status == "queued"
assert result.seconds == "6"
assert result.size == "768P"
assert result.usage == {"duration_seconds": 6.0}
def test_create_response_validates_provider_payload(self):
with pytest.raises(ValueError, match="did not return a task_id"):
self.config.transform_video_create_response(
model="MiniMax-Hailuo-2.3",
raw_response=httpx.Response(200, json={"base_resp": {"status_code": 0}}),
logging_obj=self.logging_obj,
)
with pytest.raises(Exception, match="quota exceeded"):
self.config.transform_video_create_response(
model="MiniMax-Hailuo-2.3",
raw_response=httpx.Response(
200,
json={
"base_resp": {
"status_code": 1001,
"status_msg": "quota exceeded",
}
},
),
logging_obj=self.logging_obj,
)
with pytest.raises(Exception, match="upstream error"):
self.config.transform_video_create_response(
model="MiniMax-Hailuo-2.3",
raw_response=httpx.Response(500, text="upstream error"),
logging_obj=self.logging_obj,
)
with pytest.raises(ValueError, match="invalid JSON response"):
self.config.transform_video_create_response(
model="MiniMax-Hailuo-2.3",
raw_response=httpx.Response(200, text="not-json"),
logging_obj=self.logging_obj,
)
def test_status_request_and_response(self):
encoded_id = encode_video_id_with_provider("task-123", "minimax", "MiniMax-Hailuo-2.3")
@ -107,6 +344,46 @@ class TestMinimaxVideoTransformation:
assert decoded["custom_llm_provider"] == "minimax"
assert result.status == "completed"
def test_content_request_and_failed_status_response(self):
encoded_id = encode_video_id_with_provider("task/id", "minimax", "MiniMax-Hailuo-2.3")
content_url, content_data = self.config.transform_video_content_request(
video_id=encoded_id,
api_base="https://api.minimax.io/v1/",
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert content_url == "https://api.minimax.io/v1/query/video_generation?task_id=task%2Fid"
assert content_data == {}
result = self.config.transform_video_status_retrieve_response(
raw_response=httpx.Response(
200,
json={
"task_id": "task-123",
"model": "MiniMax-Hailuo-2.3",
"status": "Failed",
"duration": 10,
"resolution": "1080P",
"base_resp": {"status_code": "0"},
},
),
logging_obj=self.logging_obj,
)
assert result.id == "task-123"
assert result.status == "failed"
assert result.error == {
"code": "generation_failed",
"message": "Failed",
}
assert result.seconds == "10"
assert result.size == "1080P"
with pytest.raises(ValueError, match="did not return a task_id"):
self.config.transform_video_status_retrieve_response(
raw_response=httpx.Response(200, json={"base_resp": {"status_code": 0}}),
logging_obj=self.logging_obj,
)
def test_content_response_retrieves_file_and_downloads_video(self):
query_request = httpx.Request(
"GET",
@ -130,15 +407,213 @@ class TestMinimaxVideoTransformation:
request=httpx.Request("GET", "https://cdn.example.com/video.mp4"),
)
client = Mock()
client.get.side_effect = [file_response, video_response]
client.get.return_value = file_response
with patch(
"litellm.llms.minimax.videos.transformation._get_httpx_client",
return_value=client,
with (
patch(
"litellm.llms.minimax.videos.transformation._get_httpx_client",
return_value=client,
),
patch(
"litellm.llms.minimax.videos.transformation.safe_get",
return_value=video_response,
) as safe_get_mock,
):
result = self.config.transform_video_content_response(query_response, self.logging_obj)
assert result == b"video-bytes"
assert client.get.call_args_list[0].args[0] == ("https://api.minimax.io/v1/files/retrieve?file_id=file-123")
assert client.get.call_args_list[0].kwargs["headers"]["Authorization"] == "Bearer test-key"
assert client.get.call_args_list[1].args[0] == "https://cdn.example.com/video.mp4"
safe_get_mock.assert_called_once_with(client, "https://cdn.example.com/video.mp4")
def test_content_response_handles_binary_and_incomplete_results(self):
query_response = httpx.Response(
200,
json={"task_id": "task-123", "status": "Success", "file_id": "file-123"},
request=httpx.Request(
"GET",
"https://api.minimax.io/v1/query/video_generation?task_id=task-123",
),
)
binary_response = httpx.Response(
200,
content=b"video-bytes",
headers={"content-type": "application/octet-stream"},
)
client = Mock()
client.get.return_value = binary_response
with patch(
"litellm.llms.minimax.videos.transformation._get_httpx_client",
return_value=client,
):
assert self.config.transform_video_content_response(query_response, self.logging_obj) == b"video-bytes"
with pytest.raises(ValueError, match="not ready for download"):
self.config.transform_video_content_response(
httpx.Response(200, json={"status": "Processing"}),
self.logging_obj,
)
client.get.return_value = httpx.Response(200, json={"base_resp": {"status_code": 0}})
with (
patch(
"litellm.llms.minimax.videos.transformation._get_httpx_client",
return_value=client,
),
pytest.raises(ValueError, match="did not return a video download URL"),
):
self.config.transform_video_content_response(query_response, self.logging_obj)
@pytest.mark.asyncio
async def test_async_content_response_downloads_video(self):
query_response = httpx.Response(
200,
json={"task_id": "task-123", "status": "Success", "file_id": "file-123"},
request=httpx.Request(
"GET",
"https://api.minimax.io/v1/query/video_generation?task_id=task-123",
headers={"Authorization": "Bearer test-key"},
),
)
file_response = httpx.Response(
200,
json={"download_url": "https://cdn.example.com/video.mp4"},
)
video_response = httpx.Response(200, content=b"async-video")
client = Mock()
client.get = AsyncMock(return_value=file_response)
with (
patch(
"litellm.llms.minimax.videos.transformation.get_async_httpx_client",
return_value=client,
),
patch(
"litellm.llms.minimax.videos.transformation.async_safe_get",
new=AsyncMock(return_value=video_response),
) as safe_get_mock,
):
result = await self.config.async_transform_video_content_response(
query_response,
self.logging_obj,
)
assert result == b"async-video"
safe_get_mock.assert_awaited_once_with(client, "https://cdn.example.com/video.mp4")
@pytest.mark.asyncio
async def test_async_content_response_handles_binary_and_incomplete_results(self):
query_response = httpx.Response(
200,
json={"file_id": "file-123"},
request=httpx.Request("GET", "https://custom.example.com/query/video_generation"),
)
client = Mock()
client.get = AsyncMock(
return_value=httpx.Response(
200,
content=b"video-bytes",
headers={"content-type": "video/mp4"},
)
)
with patch(
"litellm.llms.minimax.videos.transformation.get_async_httpx_client",
return_value=client,
):
assert (
await self.config.async_transform_video_content_response(
query_response,
self.logging_obj,
)
== b"video-bytes"
)
with pytest.raises(ValueError, match="not ready for download"):
await self.config.async_transform_video_content_response(
httpx.Response(200, json={"status": "Queued"}),
self.logging_obj,
)
client.get = AsyncMock(return_value=httpx.Response(200, json={"base_resp": {"status_code": 0}}))
with (
patch(
"litellm.llms.minimax.videos.transformation.get_async_httpx_client",
return_value=client,
),
pytest.raises(ValueError, match="did not return a video download URL"),
):
await self.config.async_transform_video_content_response(
query_response,
self.logging_obj,
)
def test_unsupported_video_operations(self):
request_args = {
"api_base": "https://api.minimax.io/v1",
"litellm_params": GenericLiteLLMParams(),
"headers": {},
}
with pytest.raises(NotImplementedError, match="remix"):
self.config.transform_video_remix_request(
video_id="task-123",
prompt="new prompt",
**request_args,
)
with pytest.raises(NotImplementedError, match="remix"):
self.config.transform_video_remix_response(
raw_response=httpx.Response(200),
logging_obj=self.logging_obj,
)
with pytest.raises(NotImplementedError, match="listing"):
self.config.transform_video_list_request(**request_args)
with pytest.raises(NotImplementedError, match="listing"):
self.config.transform_video_list_response(
raw_response=httpx.Response(200),
logging_obj=self.logging_obj,
)
with pytest.raises(NotImplementedError, match="deletion"):
self.config.transform_video_delete_request(video_id="task-123", **request_args)
with pytest.raises(NotImplementedError, match="deletion"):
self.config.transform_video_delete_response(
raw_response=httpx.Response(200),
logging_obj=self.logging_obj,
)
@pytest.mark.parametrize(
("provider_status", "openai_status"),
[
("Succeeded", "completed"),
("Canceled", "failed"),
("Preparing", "queued"),
("Processing", "in_progress"),
(None, "in_progress"),
],
)
def test_status_mapping(self, provider_status, openai_status):
assert self.config._map_status(provider_status) == openai_status
def test_response_helpers(self):
assert self.config._request_headers(httpx.Response(200)) == {}
assert self.config._api_base_from_response(httpx.Response(200)) == "https://api.minimax.io/v1"
custom_response = httpx.Response(
200,
request=httpx.Request("GET", "https://custom.example.com/query/video_generation"),
)
assert self.config._api_base_from_response(custom_response) == "https://custom.example.com/v1"
assert self.config._get_download_url({"url": "https://cdn.example.com/top.mp4"}) == (
"https://cdn.example.com/top.mp4"
)
assert self.config._get_download_url({"file": {"url": "https://cdn.example.com/file.mp4"}}) == (
"https://cdn.example.com/file.mp4"
)
assert self.config._get_download_url({}) is None
empty_video = VideoObject(id="task-123", object="video", status="queued")
assert self.config._usage_from_video(empty_video) == {}
empty_video.seconds = "invalid"
assert self.config._usage_from_video(empty_video) == {}
video_data = {}
self.config._add_request_metadata(video_data, None)
assert video_data == {}