mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-09 22:31:41 +00:00
feat(minimax): add video generation support
This commit is contained in:
octo-patch
parent
3c2264cfac
commit
74bab168c9
4 changed files with 588 additions and 0 deletions
5
litellm/llms/minimax/videos/__init__.py
Normal file
5
litellm/llms/minimax/videos/__init__.py
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
"""MiniMax video generation transformation."""
|
||||
|
||||
from .transformation import MinimaxVideoConfig
|
||||
|
||||
__all__ = ["MinimaxVideoConfig"]
|
||||
435
litellm/llms/minimax/videos/transformation.py
Normal file
435
litellm/llms/minimax/videos/transformation.py
Normal file
|
|
@ -0,0 +1,435 @@
|
|||
"""MiniMax v1 video generation transformations."""
|
||||
|
||||
from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
|
||||
from urllib.parse import quote, urlsplit, urlunsplit
|
||||
|
||||
import httpx
|
||||
from httpx._types import RequestFiles
|
||||
|
||||
import litellm
|
||||
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 (
|
||||
AsyncHTTPHandler,
|
||||
HTTPHandler,
|
||||
_get_httpx_client,
|
||||
get_async_httpx_client,
|
||||
)
|
||||
from litellm.secret_managers.main import get_secret_str
|
||||
from litellm.types.router import GenericLiteLLMParams
|
||||
from litellm.types.videos.main import VideoCreateOptionalRequestParams, VideoObject
|
||||
from litellm.types.videos.utils import (
|
||||
encode_video_id_with_provider,
|
||||
extract_original_video_id,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
|
||||
|
||||
LiteLLMLoggingObj = _LiteLLMLoggingObj
|
||||
else:
|
||||
LiteLLMLoggingObj = Any
|
||||
|
||||
|
||||
class MinimaxVideoConfig(BaseVideoConfig):
|
||||
"""Configuration for MiniMax's v1 text-to-video and image-to-video API."""
|
||||
|
||||
def get_supported_openai_params(self, model: str) -> list:
|
||||
return [
|
||||
"model",
|
||||
"prompt",
|
||||
"input_reference",
|
||||
"seconds",
|
||||
"size",
|
||||
"user",
|
||||
"extra_headers",
|
||||
"extra_body",
|
||||
"prompt_optimizer",
|
||||
"fast_pretreatment",
|
||||
"duration",
|
||||
"resolution",
|
||||
"callback_url",
|
||||
]
|
||||
|
||||
def map_openai_params(
|
||||
self,
|
||||
video_create_optional_params: VideoCreateOptionalRequestParams,
|
||||
model: str,
|
||||
drop_params: bool,
|
||||
) -> Dict:
|
||||
mapped_params: Dict[str, Any] = {}
|
||||
|
||||
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
|
||||
elif key == "seconds":
|
||||
mapped_params["duration"] = self._coerce_duration(value)
|
||||
elif key == "size":
|
||||
mapped_params["resolution"] = value
|
||||
elif key != "extra_body":
|
||||
mapped_params[key] = value
|
||||
|
||||
extra_body = video_create_optional_params.get("extra_body")
|
||||
if isinstance(extra_body, dict):
|
||||
mapped_params.update({key: value for key, value in extra_body.items() if value is not None})
|
||||
|
||||
return mapped_params
|
||||
|
||||
def validate_environment(
|
||||
self,
|
||||
headers: dict,
|
||||
model: str,
|
||||
api_key: Optional[str] = None,
|
||||
litellm_params: Optional[GenericLiteLLMParams] = None,
|
||||
) -> dict:
|
||||
if litellm_params and litellm_params.api_key:
|
||||
api_key = api_key or litellm_params.api_key
|
||||
|
||||
api_key = api_key or litellm.api_key or get_secret_str("MINIMAX_API_KEY")
|
||||
if not api_key:
|
||||
raise ValueError(
|
||||
"MiniMax API key is required. Set MINIMAX_API_KEY environment variable or pass api_key parameter."
|
||||
)
|
||||
|
||||
headers.update(
|
||||
{
|
||||
"Authorization": f"Bearer {api_key}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
)
|
||||
return headers
|
||||
|
||||
def get_complete_url(
|
||||
self,
|
||||
model: str,
|
||||
api_base: Optional[str],
|
||||
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"):
|
||||
if base_url.endswith(suffix):
|
||||
base_url = base_url[: -len(suffix)]
|
||||
break
|
||||
if not base_url.endswith("/v1"):
|
||||
base_url = f"{base_url}/v1"
|
||||
return base_url
|
||||
|
||||
def transform_video_create_request(
|
||||
self,
|
||||
model: str,
|
||||
prompt: str,
|
||||
api_base: str,
|
||||
video_create_optional_request_params: Dict,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict,
|
||||
) -> Tuple[Dict, RequestFiles, str]:
|
||||
request_data: Dict[str, Any] = {"model": model, "prompt": prompt}
|
||||
request_data.update(video_create_optional_request_params)
|
||||
request_data.pop("extra_headers", None)
|
||||
request_data.pop("extra_body", None)
|
||||
request_data.pop("user", None)
|
||||
return request_data, [], f"{api_base.rstrip('/')}/video_generation"
|
||||
|
||||
def transform_video_create_response(
|
||||
self,
|
||||
model: str,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
custom_llm_provider: Optional[str] = None,
|
||||
request_data: Optional[Dict] = None,
|
||||
) -> VideoObject:
|
||||
response_data = self._parse_json_response(raw_response)
|
||||
self._raise_for_provider_error(raw_response, response_data)
|
||||
task_id = response_data.get("task_id")
|
||||
if task_id is None:
|
||||
raise ValueError("MiniMax did not return a task_id for video generation")
|
||||
|
||||
video_data: Dict[str, Any] = {
|
||||
"id": str(task_id),
|
||||
"object": "video",
|
||||
"status": self._map_status(response_data.get("status", "queueing")),
|
||||
"model": model,
|
||||
}
|
||||
self._add_request_metadata(video_data, request_data)
|
||||
video_obj = VideoObject(**video_data)
|
||||
self._wrap_video_id(video_obj, custom_llm_provider, model)
|
||||
video_obj.usage = self._usage_from_video(video_obj)
|
||||
return video_obj
|
||||
|
||||
def transform_video_content_request(
|
||||
self,
|
||||
video_id: str,
|
||||
api_base: str,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict,
|
||||
variant: Optional[str] = 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}", {}
|
||||
|
||||
def transform_video_content_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
) -> bytes:
|
||||
query_data = self._parse_json_response(raw_response)
|
||||
self._raise_for_provider_error(raw_response, query_data)
|
||||
file_id = query_data.get("file_id")
|
||||
if file_id is None:
|
||||
status = self._map_status(query_data.get("status", "processing"))
|
||||
raise ValueError(f"MiniMax video is not ready for download (status: {status})")
|
||||
|
||||
headers = self._request_headers(raw_response)
|
||||
api_base = self._api_base_from_response(raw_response)
|
||||
file_url = f"{api_base}/files/retrieve?file_id={quote(str(file_id), safe='')}"
|
||||
client: HTTPHandler = _get_httpx_client()
|
||||
file_response = client.get(file_url, headers=headers)
|
||||
self._raise_for_status(file_response)
|
||||
if self._is_binary_response(file_response):
|
||||
return file_response.content
|
||||
|
||||
file_data = self._parse_json_response(file_response)
|
||||
self._raise_for_provider_error(file_response, file_data)
|
||||
download_url = self._get_download_url(file_data)
|
||||
if not download_url:
|
||||
raise ValueError("MiniMax did not return a video download URL")
|
||||
|
||||
video_response = client.get(download_url)
|
||||
self._raise_for_status(video_response)
|
||||
return video_response.content
|
||||
|
||||
async def async_transform_video_content_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
) -> bytes:
|
||||
query_data = self._parse_json_response(raw_response)
|
||||
self._raise_for_provider_error(raw_response, query_data)
|
||||
file_id = query_data.get("file_id")
|
||||
if file_id is None:
|
||||
status = self._map_status(query_data.get("status", "processing"))
|
||||
raise ValueError(f"MiniMax video is not ready for download (status: {status})")
|
||||
|
||||
headers = self._request_headers(raw_response)
|
||||
api_base = self._api_base_from_response(raw_response)
|
||||
file_url = f"{api_base}/files/retrieve?file_id={quote(str(file_id), safe='')}"
|
||||
client: AsyncHTTPHandler = get_async_httpx_client(llm_provider=litellm.LlmProviders.MINIMAX)
|
||||
file_response = await client.get(file_url, headers=headers)
|
||||
self._raise_for_status(file_response)
|
||||
if self._is_binary_response(file_response):
|
||||
return file_response.content
|
||||
|
||||
file_data = self._parse_json_response(file_response)
|
||||
self._raise_for_provider_error(file_response, file_data)
|
||||
download_url = self._get_download_url(file_data)
|
||||
if not download_url:
|
||||
raise ValueError("MiniMax did not return a video download URL")
|
||||
|
||||
video_response = await client.get(download_url)
|
||||
self._raise_for_status(video_response)
|
||||
return video_response.content
|
||||
|
||||
def transform_video_status_retrieve_request(
|
||||
self,
|
||||
video_id: str,
|
||||
api_base: str,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: 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}", {}
|
||||
|
||||
def transform_video_status_retrieve_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
custom_llm_provider: Optional[str] = None,
|
||||
) -> VideoObject:
|
||||
response_data = self._parse_json_response(raw_response)
|
||||
self._raise_for_provider_error(raw_response, response_data)
|
||||
task_id = response_data.get("task_id")
|
||||
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] = {
|
||||
"id": str(task_id),
|
||||
"object": "video",
|
||||
"status": self._map_status(response_data.get("status", "processing")),
|
||||
"model": model,
|
||||
}
|
||||
if response_data.get("status") and self._map_status(response_data["status"]) == "failed":
|
||||
video_data["error"] = {
|
||||
"code": "generation_failed",
|
||||
"message": str(response_data.get("status")),
|
||||
}
|
||||
if response_data.get("duration") is not None:
|
||||
video_data["seconds"] = str(response_data["duration"])
|
||||
if response_data.get("resolution") is not None:
|
||||
video_data["size"] = str(response_data["resolution"])
|
||||
|
||||
video_obj = VideoObject(**video_data)
|
||||
self._wrap_video_id(video_obj, custom_llm_provider, model)
|
||||
return video_obj
|
||||
|
||||
def transform_video_remix_request(
|
||||
self,
|
||||
video_id: str,
|
||||
prompt: str,
|
||||
api_base: str,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict,
|
||||
extra_body: Optional[Dict[str, Any]] = 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,
|
||||
) -> VideoObject:
|
||||
raise NotImplementedError("Video remix is not supported by the MiniMax v1 API")
|
||||
|
||||
def transform_video_list_request(
|
||||
self,
|
||||
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]:
|
||||
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]:
|
||||
raise NotImplementedError("Video listing is not supported by the MiniMax v1 API")
|
||||
|
||||
def transform_video_delete_request(
|
||||
self,
|
||||
video_id: str,
|
||||
api_base: str,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict,
|
||||
) -> Tuple[str, Dict]:
|
||||
raise NotImplementedError("Video deletion is not supported by the MiniMax v1 API")
|
||||
|
||||
def transform_video_delete_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
) -> 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:
|
||||
return BaseLLMException(status_code=status_code, message=error_message, headers=headers)
|
||||
|
||||
@staticmethod
|
||||
def _coerce_duration(value: Any) -> Any:
|
||||
try:
|
||||
return int(float(value))
|
||||
except (TypeError, ValueError):
|
||||
return value
|
||||
|
||||
@staticmethod
|
||||
def _map_status(status: Any) -> str:
|
||||
normalized = str(status or "").strip().lower().replace(" ", "_")
|
||||
if normalized in {"success", "succeeded", "completed", "complete"}:
|
||||
return "completed"
|
||||
if normalized in {"fail", "failed", "error", "cancelled", "canceled"}:
|
||||
return "failed"
|
||||
if normalized in {"queueing", "queued", "preparing", "pending"}:
|
||||
return "queued"
|
||||
return "in_progress"
|
||||
|
||||
@staticmethod
|
||||
def _add_request_metadata(video_data: Dict[str, Any], request_data: Optional[Dict]) -> None:
|
||||
if not request_data:
|
||||
return
|
||||
if request_data.get("duration") is not None:
|
||||
video_data["seconds"] = str(request_data["duration"])
|
||||
if request_data.get("resolution") is not None:
|
||||
video_data["size"] = str(request_data["resolution"])
|
||||
|
||||
@staticmethod
|
||||
def _usage_from_video(video_obj: VideoObject) -> Dict[str, Any]:
|
||||
if video_obj.seconds is None:
|
||||
return {}
|
||||
try:
|
||||
return {"duration_seconds": float(video_obj.seconds)}
|
||||
except (TypeError, ValueError):
|
||||
return {}
|
||||
|
||||
@staticmethod
|
||||
def _wrap_video_id(video_obj: VideoObject, provider: Optional[str], model: Optional[str]) -> 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]:
|
||||
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:
|
||||
self._raise_for_status(raw_response)
|
||||
base_resp = response_data.get("base_resp") or {}
|
||||
status_code = base_resp.get("status_code")
|
||||
if status_code not in (None, 0, "0"):
|
||||
message = base_resp.get("status_msg") or "MiniMax video request failed"
|
||||
raise self.get_error_class(str(message), raw_response.status_code, raw_response.headers)
|
||||
|
||||
def _raise_for_status(self, raw_response: httpx.Response) -> None:
|
||||
if raw_response.status_code >= 400:
|
||||
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)
|
||||
request_headers = getattr(request, "headers", None)
|
||||
if isinstance(request_headers, (dict, httpx.Headers)):
|
||||
authorization = request_headers.get("Authorization")
|
||||
if authorization:
|
||||
return {"Authorization": authorization}
|
||||
return {}
|
||||
|
||||
@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)
|
||||
if request_url is None:
|
||||
return "https://api.minimax.io/v1"
|
||||
parsed = urlsplit(str(request_url))
|
||||
path = parsed.path
|
||||
v1_index = path.find("/v1/")
|
||||
root_path = path[: v1_index + len("/v1")] if v1_index >= 0 else "/v1"
|
||||
return urlunsplit((parsed.scheme, parsed.netloc, root_path, "", ""))
|
||||
|
||||
@staticmethod
|
||||
def _is_binary_response(raw_response: httpx.Response) -> bool:
|
||||
content_type = raw_response.headers.get("content-type", "")
|
||||
return content_type.startswith("video/") or content_type == "application/octet-stream"
|
||||
|
||||
@staticmethod
|
||||
def _get_download_url(response_data: Dict[str, Any]) -> Optional[str]:
|
||||
file_data = response_data.get("file")
|
||||
if isinstance(file_data, dict):
|
||||
for key in ("download_url", "url"):
|
||||
if file_data.get(key):
|
||||
return str(file_data[key])
|
||||
for key in ("download_url", "url"):
|
||||
if response_data.get(key):
|
||||
return str(response_data[key])
|
||||
return None
|
||||
|
|
@ -8710,6 +8710,10 @@ class ProviderConfigManager:
|
|||
from litellm.llms.runwayml.videos.transformation import RunwayMLVideoConfig
|
||||
|
||||
return RunwayMLVideoConfig()
|
||||
elif LlmProviders.MINIMAX == provider:
|
||||
from litellm.llms.minimax.videos.transformation import MinimaxVideoConfig
|
||||
|
||||
return MinimaxVideoConfig()
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
|
|
|
|||
|
|
@ -0,0 +1,144 @@
|
|||
"""Tests for MiniMax v1 video generation transformations."""
|
||||
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import httpx
|
||||
|
||||
import litellm
|
||||
from litellm.llms.minimax.videos.transformation import MinimaxVideoConfig
|
||||
from litellm.types.router import GenericLiteLLMParams
|
||||
from litellm.types.videos.main import VideoObject
|
||||
from litellm.types.videos.utils import decode_video_id_with_provider, encode_video_id_with_provider
|
||||
from litellm.utils import ProviderConfigManager
|
||||
|
||||
|
||||
class TestMinimaxVideoTransformation:
|
||||
def setup_method(self):
|
||||
self.config = MinimaxVideoConfig()
|
||||
self.logging_obj = Mock()
|
||||
|
||||
def test_provider_config_is_registered(self):
|
||||
config = ProviderConfigManager.get_provider_video_config(
|
||||
model="MiniMax-Hailuo-2.3",
|
||||
provider=litellm.LlmProviders.MINIMAX,
|
||||
)
|
||||
assert isinstance(config, MinimaxVideoConfig)
|
||||
|
||||
def test_transform_create_request_maps_text_and_image_parameters(self):
|
||||
params = self.config.map_openai_params(
|
||||
{
|
||||
"input_reference": "https://example.com/frame.png",
|
||||
"seconds": "6",
|
||||
"size": "768P",
|
||||
"extra_body": {"prompt_optimizer": True},
|
||||
},
|
||||
model="MiniMax-Hailuo-2.3",
|
||||
drop_params=False,
|
||||
)
|
||||
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=params,
|
||||
litellm_params=GenericLiteLLMParams(),
|
||||
headers={},
|
||||
)
|
||||
|
||||
assert data == {
|
||||
"model": "MiniMax-Hailuo-2.3",
|
||||
"prompt": "A city at sunrise",
|
||||
"first_frame_image": "https://example.com/frame.png",
|
||||
"duration": 6,
|
||||
"resolution": "768P",
|
||||
"prompt_optimizer": True,
|
||||
}
|
||||
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,
|
||||
json={"task_id": "task-123", "base_resp": {"status_code": 0}},
|
||||
)
|
||||
result = self.config.transform_video_create_response(
|
||||
model="MiniMax-Hailuo-2.3",
|
||||
raw_response=response,
|
||||
logging_obj=self.logging_obj,
|
||||
custom_llm_provider="minimax",
|
||||
request_data={"duration": 6, "resolution": "768P"},
|
||||
)
|
||||
|
||||
assert isinstance(result, VideoObject)
|
||||
decoded = decode_video_id_with_provider(result.id)
|
||||
assert decoded["custom_llm_provider"] == "minimax"
|
||||
assert decoded["model_id"] == "MiniMax-Hailuo-2.3"
|
||||
assert decoded["video_id"] == "task-123"
|
||||
assert result.status == "queued"
|
||||
assert result.seconds == "6"
|
||||
assert result.size == "768P"
|
||||
|
||||
def test_status_request_and_response(self):
|
||||
encoded_id = encode_video_id_with_provider("task-123", "minimax", "MiniMax-Hailuo-2.3")
|
||||
url, data = self.config.transform_video_status_retrieve_request(
|
||||
video_id=encoded_id,
|
||||
api_base="https://api.minimaxi.com/v1",
|
||||
litellm_params=GenericLiteLLMParams(),
|
||||
headers={},
|
||||
)
|
||||
assert url == "https://api.minimaxi.com/v1/query/video_generation?task_id=task-123"
|
||||
assert data == {}
|
||||
|
||||
response = httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"task_id": "task-123",
|
||||
"status": "Success",
|
||||
"file_id": "file-123",
|
||||
"base_resp": {"status_code": 0},
|
||||
},
|
||||
)
|
||||
result = self.config.transform_video_status_retrieve_response(
|
||||
raw_response=response,
|
||||
logging_obj=self.logging_obj,
|
||||
custom_llm_provider="minimax",
|
||||
)
|
||||
decoded = decode_video_id_with_provider(result.id)
|
||||
assert decoded["video_id"] == "task-123"
|
||||
assert decoded["custom_llm_provider"] == "minimax"
|
||||
assert result.status == "completed"
|
||||
|
||||
def test_content_response_retrieves_file_and_downloads_video(self):
|
||||
query_request = httpx.Request(
|
||||
"GET",
|
||||
"https://api.minimax.io/v1/query/video_generation?task_id=task-123",
|
||||
headers={"Authorization": "Bearer test-key"},
|
||||
)
|
||||
query_response = httpx.Response(
|
||||
200,
|
||||
json={"task_id": "task-123", "status": "Success", "file_id": "file-123"},
|
||||
request=query_request,
|
||||
)
|
||||
file_response = httpx.Response(
|
||||
200,
|
||||
json={"file": {"download_url": "https://cdn.example.com/video.mp4"}},
|
||||
request=httpx.Request("GET", "https://api.minimax.io/v1/files/retrieve"),
|
||||
)
|
||||
video_response = httpx.Response(
|
||||
200,
|
||||
content=b"video-bytes",
|
||||
headers={"content-type": "video/mp4"},
|
||||
request=httpx.Request("GET", "https://cdn.example.com/video.mp4"),
|
||||
)
|
||||
client = Mock()
|
||||
client.get.side_effect = [file_response, video_response]
|
||||
|
||||
with patch(
|
||||
"litellm.llms.minimax.videos.transformation._get_httpx_client",
|
||||
return_value=client,
|
||||
):
|
||||
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"
|
||||
Loading…
Add table
Reference in a new issue