mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-04 02:31:27 +00:00
feat(black_forest_labs): add FLUX 3 video generation
Black Forest Labs is wired up for image generation and image editing but not video, so `black_forest_labs/flux-3-video` fails with "video generation is not supported for black_forest_labs". FLUX 3 is a single endpoint with a `mode` discriminator, so the config infers the mode from the inputs: a prompt alone is t2v, keyframes are i2v, a start_video is v2v, and a draft_cache is draft_enhance. Submission returns a regional polling URL that has to be reused, because the global host answers 404 for a job dispatched to a region.
This commit is contained in:
Maanav Dalal
parent
cde5465e6f
commit
55699fe191
6 changed files with 798 additions and 0 deletions
3
litellm/llms/black_forest_labs/videos/__init__.py
Normal file
3
litellm/llms/black_forest_labs/videos/__init__.py
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
from .transformation import BlackForestLabsVideoConfig
|
||||
|
||||
__all__ = ["BlackForestLabsVideoConfig"]
|
||||
462
litellm/llms/black_forest_labs/videos/transformation.py
Normal file
462
litellm/llms/black_forest_labs/videos/transformation.py
Normal file
|
|
@ -0,0 +1,462 @@
|
|||
"""
|
||||
Black Forest Labs FLUX 3 Video Configuration
|
||||
|
||||
Handles transformation between OpenAI-compatible video params and the Black
|
||||
Forest Labs FLUX 3 video API.
|
||||
|
||||
API Reference: https://docs.bfl.ai/api-reference/utility/generate-a-video-with-flux-3
|
||||
"""
|
||||
|
||||
import time
|
||||
from typing import TYPE_CHECKING, Any, Final # noqa: TID251 # BaseVideoConfig types its payloads dict[str, Any]
|
||||
|
||||
import httpx
|
||||
|
||||
import litellm
|
||||
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 VideoObject
|
||||
from litellm.types.videos.utils import (
|
||||
encode_video_id_with_provider,
|
||||
extract_original_video_id,
|
||||
)
|
||||
|
||||
from ..common_utils import (
|
||||
DEFAULT_API_BASE,
|
||||
BlackForestLabsError,
|
||||
assert_bfl_polling_url,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
|
||||
|
||||
LiteLLMLoggingObj = _LiteLLMLoggingObj
|
||||
else:
|
||||
LiteLLMLoggingObj = Any
|
||||
|
||||
VIDEO_MODELS: Final[dict[str, str]] = {"flux-3-video": "/v1/flux-3-video"}
|
||||
|
||||
RESOLUTIONS: Final = ("hd", "fhd")
|
||||
ASPECT_RATIOS: Final = ("21:9", "2:1", "16:9", "4:3", "1:1", "3:4", "9:16", "auto")
|
||||
MIN_DURATION: Final = 5
|
||||
MAX_DURATION: Final = 20
|
||||
|
||||
# BFL reports one of these on GET /v1/get_result. Anything outside the terminal
|
||||
# set is still running.
|
||||
_TERMINAL_STATUSES: Final[dict[str, str]] = {
|
||||
"Ready": "completed",
|
||||
"Error": "failed",
|
||||
"Content Moderated": "failed",
|
||||
"Request Moderated": "failed",
|
||||
"Task not found": "failed",
|
||||
}
|
||||
_IN_PROGRESS_STATUSES: Final[dict[str, str]] = {
|
||||
"Pending": "queued",
|
||||
"Queued": "queued",
|
||||
"Reasoning": "in_progress",
|
||||
"Generating": "in_progress",
|
||||
"Uploading": "in_progress",
|
||||
}
|
||||
|
||||
|
||||
class BlackForestLabsVideoConfig(BaseVideoConfig):
|
||||
"""
|
||||
Configuration for Black Forest Labs FLUX 3 video generation.
|
||||
|
||||
FLUX 3 is a single endpoint with a ``mode`` discriminator: ``t2v`` from a
|
||||
prompt alone, ``i2v`` from keyframe images, ``v2v`` to continue an existing
|
||||
clip, and ``draft_enhance`` to re-render a draft at full quality.
|
||||
|
||||
Submission returns a job id plus a regional ``polling_url``. That URL has to
|
||||
be reused verbatim, because the global host answers 404 for a job dispatched
|
||||
to a region.
|
||||
"""
|
||||
|
||||
def get_supported_openai_params(self, model: str) -> list:
|
||||
return [
|
||||
"model",
|
||||
"seconds",
|
||||
"size",
|
||||
"input_reference",
|
||||
"user",
|
||||
"extra_headers",
|
||||
]
|
||||
|
||||
def map_openai_params(
|
||||
self,
|
||||
video_create_optional_params: dict,
|
||||
model: str,
|
||||
drop_params: bool,
|
||||
) -> dict:
|
||||
"""
|
||||
Map OpenAI video params onto FLUX 3 params.
|
||||
|
||||
- ``seconds`` -> ``duration`` (whole seconds, 5 to 20)
|
||||
- ``size`` -> ``resolution`` tier, by the shorter side
|
||||
- ``input_reference`` -> a single ``keyframes`` entry, which selects i2v
|
||||
"""
|
||||
mapped: dict[str, Any] = {}
|
||||
|
||||
seconds = video_create_optional_params.get("seconds")
|
||||
if seconds is not None:
|
||||
duration = self._map_duration(seconds)
|
||||
if duration is not None:
|
||||
mapped["duration"] = duration
|
||||
|
||||
size = video_create_optional_params.get("size")
|
||||
if size is not None:
|
||||
resolution = self._map_size_to_resolution(size)
|
||||
if resolution is not None:
|
||||
mapped["resolution"] = resolution
|
||||
|
||||
input_reference = video_create_optional_params.get("input_reference")
|
||||
if input_reference is not None:
|
||||
mapped["keyframes"] = [input_reference]
|
||||
|
||||
supported: Final = self.get_supported_openai_params(model)
|
||||
mapped.update(
|
||||
{
|
||||
key: value
|
||||
for key, value in video_create_optional_params.items()
|
||||
if key not in supported and key not in ("seconds", "size", "input_reference")
|
||||
}
|
||||
)
|
||||
|
||||
return mapped
|
||||
|
||||
def _map_duration(self, seconds: object) -> int | None:
|
||||
if not isinstance(seconds, (int, float, str)):
|
||||
return None
|
||||
try:
|
||||
duration = int(float(seconds))
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
return max(MIN_DURATION, min(MAX_DURATION, duration))
|
||||
|
||||
def _map_size_to_resolution(self, size: object) -> str | None:
|
||||
"""
|
||||
FLUX 3 takes a named tier, not pixel dimensions, so map by the shorter
|
||||
side: at most 720 is ``hd`` and anything larger is ``fhd``.
|
||||
"""
|
||||
if not isinstance(size, str):
|
||||
return None
|
||||
if size in RESOLUTIONS:
|
||||
return size
|
||||
if "x" not in size.lower():
|
||||
return None
|
||||
try:
|
||||
width, height = (int(part) for part in size.lower().split("x", 1))
|
||||
except ValueError:
|
||||
return None
|
||||
return "hd" if min(width, height) <= 720 else "fhd"
|
||||
|
||||
def validate_environment(
|
||||
self,
|
||||
headers: dict,
|
||||
model: str,
|
||||
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
|
||||
|
||||
final_api_key: Final = (
|
||||
api_key or litellm.api_key or get_secret_str("BFL_API_KEY") or get_secret_str("BLACK_FOREST_LABS_API_KEY")
|
||||
)
|
||||
|
||||
if not final_api_key:
|
||||
raise BlackForestLabsError(
|
||||
status_code=401,
|
||||
message="BFL_API_KEY is not set. Please set it via environment variable or pass api_key parameter.",
|
||||
)
|
||||
|
||||
headers.update(
|
||||
{
|
||||
"x-key": final_api_key,
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json",
|
||||
}
|
||||
)
|
||||
return headers
|
||||
|
||||
def get_complete_url(
|
||||
self,
|
||||
model: str,
|
||||
api_base: str | None,
|
||||
litellm_params: dict,
|
||||
) -> str:
|
||||
base_url: str = api_base or get_secret_str("BFL_API_BASE") or DEFAULT_API_BASE
|
||||
return base_url.rstrip("/")
|
||||
|
||||
def _get_model_endpoint(self, model: str) -> str:
|
||||
model_name = model.lower().split("/")[-1]
|
||||
if model_name in VIDEO_MODELS:
|
||||
return VIDEO_MODELS[model_name]
|
||||
raise ValueError(f"Unknown BFL video model: {model_name}. Supported models: {list(VIDEO_MODELS.keys())}")
|
||||
|
||||
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, list, str]:
|
||||
request_data: dict[str, Any] = {"prompt": prompt}
|
||||
request_data.update(video_create_optional_request_params)
|
||||
request_data["mode"] = self._infer_mode(request_data)
|
||||
|
||||
if request_data["mode"] == "draft_enhance":
|
||||
request_data.pop("prompt", None)
|
||||
|
||||
url: Final = f"{api_base}{self._get_model_endpoint(model)}"
|
||||
return request_data, [], url
|
||||
|
||||
def _infer_mode(self, request_data: dict) -> str:
|
||||
"""FLUX 3 discriminates on ``mode``; derive it from the inputs given."""
|
||||
if request_data.get("mode"):
|
||||
return str(request_data["mode"])
|
||||
if request_data.get("draft_cache"):
|
||||
return "draft_enhance"
|
||||
if request_data.get("start_video"):
|
||||
return "v2v"
|
||||
if request_data.get("keyframes"):
|
||||
return "i2v"
|
||||
return "t2v"
|
||||
|
||||
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,
|
||||
) -> VideoObject:
|
||||
"""
|
||||
Submission answers with the job handle, not the finished video:
|
||||
``{"id": ..., "polling_url": ..., "cost": null}``.
|
||||
|
||||
The regional polling URL is kept on the video id so status and content
|
||||
calls hit the same region.
|
||||
"""
|
||||
response_data: Final = self._parse_json(raw_response)
|
||||
|
||||
job_id: Final = response_data.get("id")
|
||||
if not job_id:
|
||||
raise BlackForestLabsError(
|
||||
status_code=raw_response.status_code,
|
||||
message=f"No job id in BFL response: {response_data}",
|
||||
)
|
||||
|
||||
polling_url: Final = response_data.get("polling_url")
|
||||
if polling_url:
|
||||
assert_bfl_polling_url(polling_url)
|
||||
|
||||
video_obj: Final = VideoObject(
|
||||
id=job_id,
|
||||
object="video",
|
||||
status="queued",
|
||||
created_at=int(time.time()),
|
||||
model=model,
|
||||
)
|
||||
|
||||
if request_data:
|
||||
if request_data.get("duration") not in (None, "auto"):
|
||||
video_obj.seconds = str(request_data["duration"])
|
||||
if request_data.get("resolution"):
|
||||
video_obj.size = str(request_data["resolution"])
|
||||
|
||||
video_obj._hidden_params = {"polling_url": polling_url}
|
||||
|
||||
if custom_llm_provider:
|
||||
video_obj.id = encode_video_id_with_provider(job_id, custom_llm_provider, model)
|
||||
|
||||
return video_obj
|
||||
|
||||
def transform_video_status_retrieve_request(
|
||||
self,
|
||||
video_id: str,
|
||||
api_base: str,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict,
|
||||
) -> tuple[str, dict]:
|
||||
return self._get_result_url(video_id, api_base), {}
|
||||
|
||||
def transform_video_status_retrieve_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
custom_llm_provider: str | None = None,
|
||||
) -> VideoObject:
|
||||
response_data: Final = self._parse_json(raw_response)
|
||||
bfl_status: Final = response_data.get("status", "Pending")
|
||||
|
||||
video_obj: Final = VideoObject(
|
||||
id=response_data.get("id", ""),
|
||||
object="video",
|
||||
status=self._map_status(bfl_status),
|
||||
progress=self._map_progress(response_data.get("progress")),
|
||||
)
|
||||
|
||||
if bfl_status in _TERMINAL_STATUSES and _TERMINAL_STATUSES[bfl_status] == "failed":
|
||||
video_obj.error = {
|
||||
"code": bfl_status,
|
||||
"message": str(response_data.get("details") or bfl_status),
|
||||
}
|
||||
|
||||
cost: Final = response_data.get("cost")
|
||||
if cost is not None:
|
||||
video_obj.usage = {"credits": cost}
|
||||
|
||||
if custom_llm_provider and video_obj.id:
|
||||
video_obj.id = encode_video_id_with_provider(video_obj.id, custom_llm_provider, None)
|
||||
|
||||
return video_obj
|
||||
|
||||
def _map_status(self, bfl_status: str) -> str:
|
||||
if bfl_status in _TERMINAL_STATUSES:
|
||||
return _TERMINAL_STATUSES[bfl_status]
|
||||
return _IN_PROGRESS_STATUSES.get(bfl_status, "in_progress")
|
||||
|
||||
def _map_progress(self, progress: object) -> int | None:
|
||||
if not isinstance(progress, (int, float, str)):
|
||||
return None
|
||||
try:
|
||||
value = float(progress)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
# BFL reports a 0..1 fraction; VideoObject.progress is a percentage.
|
||||
return round(value * 100) if value <= 1 else round(value)
|
||||
|
||||
def transform_video_content_request(
|
||||
self,
|
||||
video_id: str,
|
||||
api_base: str,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict,
|
||||
variant: str | None = None,
|
||||
) -> tuple[str, dict]:
|
||||
return self._get_result_url(video_id, api_base), {}
|
||||
|
||||
def transform_video_content_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
) -> bytes:
|
||||
video_url: Final = self._extract_video_url(self._parse_json(raw_response))
|
||||
httpx_client: Final[HTTPHandler] = _get_httpx_client()
|
||||
video_response: Final = httpx_client.get(video_url)
|
||||
video_response.raise_for_status()
|
||||
return video_response.content
|
||||
|
||||
async def async_transform_video_content_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
) -> bytes:
|
||||
video_url: Final = self._extract_video_url(self._parse_json(raw_response))
|
||||
async_client: Final[AsyncHTTPHandler] = get_async_httpx_client(
|
||||
llm_provider=litellm.LlmProviders.BLACK_FOREST_LABS,
|
||||
)
|
||||
video_response: Final = await async_client.get(video_url)
|
||||
video_response.raise_for_status()
|
||||
return video_response.content
|
||||
|
||||
def _extract_video_url(self, response_data: dict) -> str:
|
||||
status: Final = response_data.get("status", "Pending")
|
||||
result: Final = response_data.get("result") or {}
|
||||
video_url: Final = result.get("sample")
|
||||
|
||||
if video_url:
|
||||
return video_url
|
||||
|
||||
if status in _TERMINAL_STATUSES:
|
||||
raise BlackForestLabsError(
|
||||
status_code=500,
|
||||
message=f"Video generation did not produce a video (status: {status}).",
|
||||
)
|
||||
raise BlackForestLabsError(
|
||||
status_code=409,
|
||||
message=f"Video is still processing (status: {status}). Please wait and try again.",
|
||||
)
|
||||
|
||||
def _get_result_url(self, video_id: str, api_base: str) -> str:
|
||||
original_video_id: Final = extract_original_video_id(video_id)
|
||||
return f"{api_base.rstrip('/')}/v1/get_result?id={original_video_id}"
|
||||
|
||||
def _parse_json(self, raw_response: httpx.Response) -> dict:
|
||||
try:
|
||||
return raw_response.json()
|
||||
except Exception as e:
|
||||
raise BlackForestLabsError(
|
||||
status_code=raw_response.status_code,
|
||||
message=f"Error parsing BFL response: {e}",
|
||||
)
|
||||
|
||||
def transform_video_remix_request(
|
||||
self,
|
||||
video_id: str,
|
||||
prompt: str,
|
||||
api_base: str,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict,
|
||||
extra_body: dict[str, Any] | None = None,
|
||||
) -> tuple[str, dict]:
|
||||
raise NotImplementedError("video remix is not supported by the FLUX 3 video 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 FLUX 3 video API")
|
||||
|
||||
def transform_video_list_request(
|
||||
self,
|
||||
api_base: str,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict,
|
||||
after: str | None = None,
|
||||
limit: int | None = None,
|
||||
order: str | None = None,
|
||||
extra_query: dict[str, Any] | None = None,
|
||||
) -> tuple[str, dict]:
|
||||
raise NotImplementedError("video listing is not supported by the FLUX 3 video API")
|
||||
|
||||
def transform_video_list_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
custom_llm_provider: str | None = None,
|
||||
) -> dict[str, str]:
|
||||
raise NotImplementedError("video listing is not supported by the FLUX 3 video API")
|
||||
|
||||
def transform_video_delete_request(
|
||||
self,
|
||||
video_id: str,
|
||||
api_base: str,
|
||||
litellm_params: GenericLiteLLMParams,
|
||||
headers: dict,
|
||||
) -> tuple[str, dict]:
|
||||
raise NotImplementedError("video delete is not supported by the FLUX 3 video API")
|
||||
|
||||
def transform_video_delete_response(
|
||||
self,
|
||||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
) -> VideoObject:
|
||||
raise NotImplementedError("video delete is not supported by the FLUX 3 video API")
|
||||
|
||||
def get_error_class(
|
||||
self, error_message: str, status_code: int, headers: dict | httpx.Headers
|
||||
) -> BlackForestLabsError:
|
||||
return BlackForestLabsError(status_code=status_code, message=error_message)
|
||||
|
|
@ -8885,6 +8885,12 @@ class ProviderConfigManager:
|
|||
from litellm.llms.runwayml.videos.transformation import RunwayMLVideoConfig
|
||||
|
||||
return RunwayMLVideoConfig()
|
||||
elif LlmProviders.BLACK_FOREST_LABS == provider:
|
||||
from litellm.llms.black_forest_labs.videos.transformation import (
|
||||
BlackForestLabsVideoConfig,
|
||||
)
|
||||
|
||||
return BlackForestLabsVideoConfig()
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
|
|
|
|||
|
|
@ -11872,6 +11872,23 @@
|
|||
"/v1/images/generations"
|
||||
]
|
||||
},
|
||||
"black_forest_labs/flux-3-video": {
|
||||
"litellm_provider": "black_forest_labs",
|
||||
"mode": "video_generation",
|
||||
"output_cost_per_video_per_second": 0.17,
|
||||
"source": "https://docs.bfl.ai/quick_start/pricing",
|
||||
"supported_modalities": [
|
||||
"text",
|
||||
"image",
|
||||
"video"
|
||||
],
|
||||
"supported_output_modalities": [
|
||||
"video"
|
||||
],
|
||||
"metadata": {
|
||||
"comment": "Per-second pricing varies by mode and resolution: t2v and i2v are $0.17/s hd and $0.29/s fhd, v2v is $0.43/s hd and $0.54/s fhd, drafts are $0.06/s (t2v, i2v) and $0.12/s (v2v). The t2v hd rate is used here."
|
||||
}
|
||||
},
|
||||
"cerebras/llama-3.3-70b": {
|
||||
"input_cost_per_token": 8.5e-07,
|
||||
"litellm_provider": "cerebras",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,310 @@
|
|||
"""
|
||||
Tests for Black Forest Labs FLUX 3 video generation transformation.
|
||||
|
||||
Payload and response shapes are taken from a live FLUX 3 video generation
|
||||
against https://api.bfl.ai/v1/flux-3-video and its regional polling URL.
|
||||
"""
|
||||
|
||||
from unittest.mock import Mock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from litellm.llms.black_forest_labs.common_utils import BlackForestLabsError
|
||||
from litellm.llms.black_forest_labs.videos.transformation import (
|
||||
BlackForestLabsVideoConfig,
|
||||
)
|
||||
from litellm.types.router import GenericLiteLLMParams
|
||||
from litellm.types.videos.utils import extract_original_video_id
|
||||
|
||||
API_BASE = "https://api.bfl.ai"
|
||||
JOB_ID = "90307d3a-deec-47cb-bdb9-bf1c5bae1f04"
|
||||
POLLING_URL = f"https://api.us7.bfl.ai/v1/get_result?id={JOB_ID}"
|
||||
SAMPLE_URL = "https://delivery.us7.bfl.ai/durable/2026081720/video.mp4?se=2026-08-17T21%3A26%3A38Z&sig=abc"
|
||||
|
||||
|
||||
def _response(payload: dict, status_code: int = 200) -> httpx.Response:
|
||||
return httpx.Response(
|
||||
status_code=status_code,
|
||||
json=payload,
|
||||
request=httpx.Request("POST", f"{API_BASE}/v1/flux-3-video"),
|
||||
)
|
||||
|
||||
|
||||
class TestBlackForestLabsVideoTransformation:
|
||||
def setup_method(self):
|
||||
self.config = BlackForestLabsVideoConfig()
|
||||
self.mock_logging_obj = Mock()
|
||||
|
||||
def test_text_to_video_request_sets_t2v_mode(self):
|
||||
data, files, url = self.config.transform_video_create_request(
|
||||
model="flux-3-video",
|
||||
prompt="A white kitten chases a butterfly across a sunlit garden.",
|
||||
api_base=API_BASE,
|
||||
video_create_optional_request_params={
|
||||
"duration": 8,
|
||||
"resolution": "fhd",
|
||||
"aspect_ratio": "16:9",
|
||||
},
|
||||
litellm_params=GenericLiteLLMParams(),
|
||||
headers={},
|
||||
)
|
||||
|
||||
assert url == "https://api.bfl.ai/v1/flux-3-video"
|
||||
assert data["mode"] == "t2v"
|
||||
assert data["prompt"] == "A white kitten chases a butterfly across a sunlit garden."
|
||||
assert data["duration"] == 8
|
||||
assert data["resolution"] == "fhd"
|
||||
assert files == []
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params, expected_mode",
|
||||
[
|
||||
({}, "t2v"),
|
||||
({"keyframes": ["https://example.com/first.png"]}, "i2v"),
|
||||
({"start_video": "https://example.com/clip.mp4"}, "v2v"),
|
||||
({"draft_cache": "https://example.com/bundle.bin"}, "draft_enhance"),
|
||||
],
|
||||
)
|
||||
def test_mode_is_inferred_from_inputs(self, params, expected_mode):
|
||||
data, _, _ = self.config.transform_video_create_request(
|
||||
model="flux-3-video",
|
||||
prompt="a prompt",
|
||||
api_base=API_BASE,
|
||||
video_create_optional_request_params=dict(params),
|
||||
litellm_params=GenericLiteLLMParams(),
|
||||
headers={},
|
||||
)
|
||||
|
||||
assert data["mode"] == expected_mode
|
||||
|
||||
def test_draft_enhance_drops_the_prompt(self):
|
||||
"""FLUX 3 rejects a prompt in draft_enhance mode; the bundle carries it."""
|
||||
data, _, _ = self.config.transform_video_create_request(
|
||||
model="flux-3-video",
|
||||
prompt="a prompt the API would reject here",
|
||||
api_base=API_BASE,
|
||||
video_create_optional_request_params={"draft_cache": "bundle"},
|
||||
litellm_params=GenericLiteLLMParams(),
|
||||
headers={},
|
||||
)
|
||||
|
||||
assert data["mode"] == "draft_enhance"
|
||||
assert "prompt" not in data
|
||||
|
||||
def test_explicit_mode_is_not_overridden(self):
|
||||
data, _, _ = self.config.transform_video_create_request(
|
||||
model="flux-3-video",
|
||||
prompt="a prompt",
|
||||
api_base=API_BASE,
|
||||
video_create_optional_request_params={
|
||||
"mode": "i2v",
|
||||
"keyframes": ["https://example.com/first.png"],
|
||||
},
|
||||
litellm_params=GenericLiteLLMParams(),
|
||||
headers={},
|
||||
)
|
||||
|
||||
assert data["mode"] == "i2v"
|
||||
|
||||
def test_unknown_model_is_rejected(self):
|
||||
with pytest.raises(ValueError, match="Unknown BFL video model"):
|
||||
self.config.transform_video_create_request(
|
||||
model="flux-9-video",
|
||||
prompt="a prompt",
|
||||
api_base=API_BASE,
|
||||
video_create_optional_request_params={},
|
||||
litellm_params=GenericLiteLLMParams(),
|
||||
headers={},
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"size, expected_resolution",
|
||||
[
|
||||
("1280x720", "hd"),
|
||||
("1920x1080", "fhd"),
|
||||
("1080x1920", "fhd"),
|
||||
("720x1280", "hd"),
|
||||
("fhd", "fhd"),
|
||||
],
|
||||
)
|
||||
def test_size_maps_to_a_resolution_tier(self, size, expected_resolution):
|
||||
mapped = self.config.map_openai_params(
|
||||
video_create_optional_params={"size": size},
|
||||
model="flux-3-video",
|
||||
drop_params=False,
|
||||
)
|
||||
|
||||
assert mapped["resolution"] == expected_resolution
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"seconds, expected_duration",
|
||||
[("8", 8), (8, 8), ("2", 5), ("60", 20), ("8.6", 8)],
|
||||
)
|
||||
def test_seconds_maps_into_the_supported_duration_range(self, seconds, expected_duration):
|
||||
mapped = self.config.map_openai_params(
|
||||
video_create_optional_params={"seconds": seconds},
|
||||
model="flux-3-video",
|
||||
drop_params=False,
|
||||
)
|
||||
|
||||
assert mapped["duration"] == expected_duration
|
||||
|
||||
def test_input_reference_becomes_a_keyframe(self):
|
||||
mapped = self.config.map_openai_params(
|
||||
video_create_optional_params={"input_reference": "https://example.com/first.png"},
|
||||
model="flux-3-video",
|
||||
drop_params=False,
|
||||
)
|
||||
|
||||
assert mapped["keyframes"] == ["https://example.com/first.png"]
|
||||
|
||||
def test_create_response_returns_the_job_handle_and_keeps_the_polling_url(self):
|
||||
video = self.config.transform_video_create_response(
|
||||
model="flux-3-video",
|
||||
raw_response=_response({"id": JOB_ID, "polling_url": POLLING_URL, "cost": None}),
|
||||
logging_obj=self.mock_logging_obj,
|
||||
custom_llm_provider="black_forest_labs",
|
||||
request_data={"duration": 8, "resolution": "fhd"},
|
||||
)
|
||||
|
||||
assert extract_original_video_id(video.id) == JOB_ID
|
||||
assert video.status == "queued"
|
||||
assert video.seconds == "8"
|
||||
assert video.size == "fhd"
|
||||
assert video._hidden_params["polling_url"] == POLLING_URL
|
||||
|
||||
def test_create_response_without_a_job_id_raises(self):
|
||||
with pytest.raises(BlackForestLabsError):
|
||||
self.config.transform_video_create_response(
|
||||
model="flux-3-video",
|
||||
raw_response=_response({"detail": "bad request"}, status_code=422),
|
||||
logging_obj=self.mock_logging_obj,
|
||||
)
|
||||
|
||||
def test_create_response_rejects_a_polling_url_outside_bfl(self):
|
||||
with pytest.raises(BlackForestLabsError, match="not within the bfl.ai domain"):
|
||||
self.config.transform_video_create_response(
|
||||
model="flux-3-video",
|
||||
raw_response=_response(
|
||||
{"id": JOB_ID, "polling_url": "https://attacker.example.com/v1/get_result"}
|
||||
),
|
||||
logging_obj=self.mock_logging_obj,
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"bfl_status, expected_status",
|
||||
[
|
||||
("Pending", "queued"),
|
||||
("Queued", "queued"),
|
||||
("Reasoning", "in_progress"),
|
||||
("Generating", "in_progress"),
|
||||
("Ready", "completed"),
|
||||
("Error", "failed"),
|
||||
("Content Moderated", "failed"),
|
||||
("Task not found", "failed"),
|
||||
],
|
||||
)
|
||||
def test_bfl_status_maps_to_openai_status(self, bfl_status, expected_status):
|
||||
video = self.config.transform_video_status_retrieve_response(
|
||||
raw_response=_response({"id": JOB_ID, "status": bfl_status, "result": {}}),
|
||||
logging_obj=self.mock_logging_obj,
|
||||
)
|
||||
|
||||
assert video.status == expected_status
|
||||
|
||||
def test_failed_status_carries_the_bfl_detail(self):
|
||||
video = self.config.transform_video_status_retrieve_response(
|
||||
raw_response=_response(
|
||||
{"id": JOB_ID, "status": "Content Moderated", "details": "flagged by moderation"}
|
||||
),
|
||||
logging_obj=self.mock_logging_obj,
|
||||
)
|
||||
|
||||
assert video.status == "failed"
|
||||
assert video.error["code"] == "Content Moderated"
|
||||
assert video.error["message"] == "flagged by moderation"
|
||||
|
||||
def test_completed_status_has_no_error(self):
|
||||
video = self.config.transform_video_status_retrieve_response(
|
||||
raw_response=_response({"id": JOB_ID, "status": "Ready", "result": {"sample": SAMPLE_URL}}),
|
||||
logging_obj=self.mock_logging_obj,
|
||||
)
|
||||
|
||||
assert video.error is None
|
||||
|
||||
def test_credit_cost_is_reported_as_usage(self):
|
||||
video = self.config.transform_video_status_retrieve_response(
|
||||
raw_response=_response({"id": JOB_ID, "status": "Ready", "cost": 30.0}),
|
||||
logging_obj=self.mock_logging_obj,
|
||||
)
|
||||
|
||||
assert video.usage == {"credits": 30.0}
|
||||
|
||||
def test_fractional_progress_is_reported_as_a_percentage(self):
|
||||
video = self.config.transform_video_status_retrieve_response(
|
||||
raw_response=_response({"id": JOB_ID, "status": "Generating", "progress": 0.42}),
|
||||
logging_obj=self.mock_logging_obj,
|
||||
)
|
||||
|
||||
assert video.progress == 42
|
||||
|
||||
def test_status_request_targets_get_result_with_the_original_job_id(self):
|
||||
encoded_id = self.config.transform_video_create_response(
|
||||
model="flux-3-video",
|
||||
raw_response=_response({"id": JOB_ID, "polling_url": POLLING_URL}),
|
||||
logging_obj=self.mock_logging_obj,
|
||||
custom_llm_provider="black_forest_labs",
|
||||
).id
|
||||
|
||||
url, params = self.config.transform_video_status_retrieve_request(
|
||||
video_id=encoded_id,
|
||||
api_base="https://api.us7.bfl.ai",
|
||||
litellm_params=GenericLiteLLMParams(),
|
||||
headers={},
|
||||
)
|
||||
|
||||
assert url == POLLING_URL
|
||||
assert params == {}
|
||||
|
||||
def test_content_request_targets_get_result(self):
|
||||
url, _ = self.config.transform_video_content_request(
|
||||
video_id=JOB_ID,
|
||||
api_base="https://api.us7.bfl.ai",
|
||||
litellm_params=GenericLiteLLMParams(),
|
||||
headers={},
|
||||
)
|
||||
|
||||
assert url == POLLING_URL
|
||||
|
||||
def test_content_response_raises_while_still_generating(self):
|
||||
with pytest.raises(BlackForestLabsError, match="still processing"):
|
||||
self.config.transform_video_content_response(
|
||||
raw_response=_response({"id": JOB_ID, "status": "Generating", "result": {}}),
|
||||
logging_obj=self.mock_logging_obj,
|
||||
)
|
||||
|
||||
def test_content_response_raises_when_a_terminal_job_has_no_video(self):
|
||||
with pytest.raises(BlackForestLabsError, match="did not produce a video"):
|
||||
self.config.transform_video_content_response(
|
||||
raw_response=_response({"id": JOB_ID, "status": "Error", "result": {}}),
|
||||
logging_obj=self.mock_logging_obj,
|
||||
)
|
||||
|
||||
def test_validate_environment_sets_the_x_key_header(self):
|
||||
headers = self.config.validate_environment(
|
||||
headers={},
|
||||
model="flux-3-video",
|
||||
api_key="test-key",
|
||||
)
|
||||
|
||||
assert headers["x-key"] == "test-key"
|
||||
assert headers["Content-Type"] == "application/json"
|
||||
|
||||
def test_validate_environment_without_a_key_raises(self, monkeypatch):
|
||||
monkeypatch.delenv("BFL_API_KEY", raising=False)
|
||||
monkeypatch.delenv("BLACK_FOREST_LABS_API_KEY", raising=False)
|
||||
monkeypatch.setattr("litellm.api_key", None)
|
||||
|
||||
with pytest.raises(BlackForestLabsError, match="BFL_API_KEY is not set"):
|
||||
self.config.validate_environment(headers={}, model="flux-3-video")
|
||||
Loading…
Add table
Reference in a new issue