mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-06 02:48:13 +00:00
Merge b0aea9bc5b into ec94a1f82a
This commit is contained in:
Octopus
GitHub
commit
0fb5c81613
4 changed files with 1133 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",)
|
||||
505
litellm/llms/minimax/videos/transformation.py
Normal file
505
litellm/llms/minimax/videos/transformation.py
Normal file
|
|
@ -0,0 +1,505 @@
|
|||
"""MiniMax v1 video generation transformations."""
|
||||
|
||||
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 (
|
||||
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 = object
|
||||
|
||||
|
||||
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: # mutable-ok: BaseVideoConfig requires a list.
|
||||
return [ # mutable-ok: BaseVideoConfig requires a list result.
|
||||
"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: # mutable-ok: BaseVideoConfig requires a mutable parameter mapping.
|
||||
mapped_params: dict = {} # mutable-ok: Provider parameters are assembled incrementally.
|
||||
|
||||
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"] = self._prepare_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( # mutable-ok: Provider extra fields must merge into the request mapping.
|
||||
{ # mutable-ok: Provider extra fields must merge into the request mapping.
|
||||
key: value for key, value in extra_body.items() if value is not None
|
||||
}
|
||||
)
|
||||
|
||||
return mapped_params
|
||||
|
||||
def validate_environment(
|
||||
self,
|
||||
headers: dict, # mutable-ok: BaseVideoConfig passes a mutable header mapping.
|
||||
model: str,
|
||||
api_key: str | None = None,
|
||||
litellm_params: GenericLiteLLMParams | None = None,
|
||||
) -> dict: # mutable-ok: BaseVideoConfig requires mutable headers.
|
||||
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(
|
||||
{ # mutable-ok: Headers are updated in place by the provider interface.
|
||||
"Authorization": f"Bearer {api_key}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
)
|
||||
return headers
|
||||
|
||||
def get_complete_url(
|
||||
self,
|
||||
model: str,
|
||||
api_base: str | None,
|
||||
litellm_params: dict, # mutable-ok: BaseVideoConfig defines this parameter as a 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 ("/query/video_generation", "/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, # mutable-ok: BaseVideoConfig defines mutable request parameters.
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict, # mutable-ok: BaseVideoConfig passes a mutable header mapping.
|
||||
) -> tuple[dict, RequestFiles, str]: # mutable-ok: BaseVideoConfig requires a dict payload.
|
||||
request_data: dict = { # mutable-ok: Provider request fields are assembled incrementally.
|
||||
"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" # mutable-ok: No multipart files are sent.
|
||||
|
||||
def transform_video_create_response(
|
||||
self,
|
||||
model: str,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
custom_llm_provider: str | None = None,
|
||||
request_data: dict | None = None, # mutable-ok: BaseVideoConfig supplies request metadata as a dict.
|
||||
) -> 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 = { # mutable-ok: Optional response metadata is added after validation.
|
||||
"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, # mutable-ok: BaseVideoConfig passes a mutable header mapping.
|
||||
variant: str | None = None,
|
||||
) -> tuple[str, dict]: # mutable-ok: BaseVideoConfig requires a dict query payload.
|
||||
task_id = quote(extract_original_video_id(video_id), safe="")
|
||||
return ( # mutable-ok: The provider query has no separate parameter payload.
|
||||
f"{api_base.rstrip('/')}/query/video_generation?task_id={task_id}",
|
||||
{}, # mutable-ok: The provider query has no separate parameter payload.
|
||||
)
|
||||
|
||||
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 = safe_get(client, 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 async_safe_get(client, 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, # mutable-ok: BaseVideoConfig passes a mutable header mapping.
|
||||
) -> tuple[str, dict]: # mutable-ok: BaseVideoConfig requires a dict query payload.
|
||||
task_id = quote(extract_original_video_id(video_id), safe="")
|
||||
return ( # mutable-ok: The provider query has no separate parameter payload.
|
||||
f"{api_base.rstrip('/')}/query/video_generation?task_id={task_id}",
|
||||
{}, # mutable-ok: The provider query has no separate parameter payload.
|
||||
)
|
||||
|
||||
def transform_video_status_retrieve_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
custom_llm_provider: str | None = 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 = { # mutable-ok: Optional status fields are added after validation.
|
||||
"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"] = { # mutable-ok: VideoObject expects a mutable error payload.
|
||||
"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, # mutable-ok: BaseVideoConfig passes a mutable header mapping.
|
||||
extra_body: dict | None = None, # mutable-ok: BaseVideoConfig defines provider extras as a dict.
|
||||
) -> tuple[str, dict]: # mutable-ok: BaseVideoConfig requires a dict payload.
|
||||
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: str | None = 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, # mutable-ok: BaseVideoConfig passes a mutable header mapping.
|
||||
after: str | None = None,
|
||||
limit: int | None = None,
|
||||
order: str | None = None,
|
||||
extra_query: dict | None = None, # mutable-ok: BaseVideoConfig defines query extras as a dict.
|
||||
) -> tuple[str, dict]: # mutable-ok: BaseVideoConfig requires a dict payload.
|
||||
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: str | None = None,
|
||||
) -> dict[str, str]: # mutable-ok: BaseVideoConfig requires a dict response.
|
||||
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, # mutable-ok: BaseVideoConfig passes a mutable header mapping.
|
||||
) -> tuple[str, dict]: # mutable-ok: BaseVideoConfig requires a dict payload.
|
||||
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: dict | httpx.Headers, # mutable-ok: BaseLLMException accepts the response header mapping.
|
||||
) -> BaseLLMException:
|
||||
return BaseLLMException(status_code=status_code, message=error_message, headers=headers)
|
||||
|
||||
@staticmethod
|
||||
def _coerce_duration(value: object) -> object:
|
||||
try:
|
||||
return int(float(value))
|
||||
except (TypeError, ValueError):
|
||||
return value
|
||||
|
||||
@staticmethod
|
||||
def _map_status(status: object) -> 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, # mutable-ok: VideoObject metadata is assembled in place.
|
||||
request_data: dict | None, # mutable-ok: Request metadata arrives as a 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: # mutable-ok: VideoObject usage requires a dict.
|
||||
if video_obj.seconds is None:
|
||||
return {} # mutable-ok: VideoObject usage requires a dict.
|
||||
try:
|
||||
return { # mutable-ok: VideoObject usage requires a dict.
|
||||
"duration_seconds": float(video_obj.seconds)
|
||||
}
|
||||
except (TypeError, ValueError):
|
||||
return {} # mutable-ok: VideoObject usage requires a dict.
|
||||
|
||||
@staticmethod
|
||||
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)
|
||||
|
||||
def _parse_json_response(
|
||||
self, raw_response: httpx.Response
|
||||
) -> dict: # mutable-ok: JSON objects are provider dicts.
|
||||
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, # mutable-ok: Parsed provider JSON is represented as a dict.
|
||||
) -> None:
|
||||
self._raise_for_status(raw_response)
|
||||
base_resp = response_data.get("base_resp") or {} # mutable-ok: Missing provider metadata uses an empty dict.
|
||||
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]: # mutable-ok: HTTP handlers require a dict of headers.
|
||||
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} # mutable-ok: HTTP handlers require a dict of headers.
|
||||
return {} # mutable-ok: HTTP handlers require a dict of headers.
|
||||
|
||||
@staticmethod
|
||||
def _api_base_from_response(raw_response: httpx.Response) -> str:
|
||||
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))
|
||||
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, # mutable-ok: Parsed provider JSON is represented as a dict.
|
||||
) -> str | None:
|
||||
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
|
||||
|
||||
@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}"
|
||||
|
|
@ -9061,6 +9061,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,619 @@
|
|||
"""Tests for MiniMax v1 video generation transformations."""
|
||||
|
||||
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
|
||||
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_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(
|
||||
{
|
||||
"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_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,
|
||||
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"
|
||||
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")
|
||||
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_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",
|
||||
"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.return_value = file_response
|
||||
|
||||
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"
|
||||
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 == {}
|
||||
Loading…
Add table
Reference in a new issue