diff --git a/litellm/llms/black_forest_labs/videos/__init__.py b/litellm/llms/black_forest_labs/videos/__init__.py new file mode 100644 index 00000000000..97380ade709 --- /dev/null +++ b/litellm/llms/black_forest_labs/videos/__init__.py @@ -0,0 +1,3 @@ +from .transformation import BlackForestLabsVideoConfig + +__all__ = ["BlackForestLabsVideoConfig"] diff --git a/litellm/llms/black_forest_labs/videos/transformation.py b/litellm/llms/black_forest_labs/videos/transformation.py new file mode 100644 index 00000000000..78a5c2c630e --- /dev/null +++ b/litellm/llms/black_forest_labs/videos/transformation.py @@ -0,0 +1,540 @@ +""" +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 collections.abc import Mapping +from types import MappingProxyType +from typing import TYPE_CHECKING, Any, Final # noqa: TID251 # BaseVideoConfig types its payloads dict[str, Any] +from urllib.parse import urlparse + +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, # pyright: ignore[reportPrivateUsage] # the shared client factory every provider uses to fetch generated media + 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, +) + +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 # rebind-ok: the TYPE_CHECKING alias for the runtime Any below +else: + LiteLLMLoggingObj = Any # rebind-ok: runtime stand-in for the type-only Logging alias + +VIDEO_MODELS: Final[Mapping[str, str]] = MappingProxyType({"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[Mapping[str, str]] = MappingProxyType( + { + "Ready": "completed", + "Error": "failed", + "Content Moderated": "failed", + "Request Moderated": "failed", + "Task not found": "failed", + } +) +_IN_PROGRESS_STATUSES: Final[Mapping[str, str]] = MappingProxyType( + { + "Pending": "queued", + "Queued": "queued", + "Reasoning": "in_progress", + "Generating": "in_progress", + "Uploading": "in_progress", + } +) + +# BFL dispatches each job to a regional host and hands that host back in +# ``polling_url``. The global host answers 404 for a regional job, so the region +# has to survive from submission through to status and content retrieval. The +# only value carried across those calls is the video id, so the region travels +# inside it, behind a separator BFL's own UUIDs never contain. +_REGION_SEPARATOR: Final = "@" + + +def _pack_region(job_id: str, polling_url: str | None) -> str: + """Attach the polling host to a job id, when it differs from the default.""" + if not polling_url: + return job_id + host: Final = (urlparse(polling_url).hostname or "").lower() + if not host or host == urlparse(DEFAULT_API_BASE).hostname: + return job_id + return f"{job_id}{_REGION_SEPARATOR}{host}" + + +def _unpack_region(job_id: str) -> tuple[str, str | None]: + """Split a packed id back into the bare job id and its polling host.""" + if _REGION_SEPARATOR not in job_id: + return job_id, None + bare, _, host = job_id.partition(_REGION_SEPARATOR) + return bare, host or None + + +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: # mutable-ok: BaseVideoConfig signature + return [ # mutable-ok: BaseVideoConfig signature + "model", + "seconds", + "size", + "input_reference", + "user", + "extra_headers", + ] + + def map_openai_params( + self, + video_create_optional_params: VideoCreateOptionalRequestParams, + model: str, + drop_params: bool, + ) -> dict: # mutable-ok: BaseVideoConfig signature + """ + 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 + """ + supported: Final = self.get_supported_openai_params(model) + remapped: Final = ("seconds", "size", "input_reference") + + duration: Final = self._map_duration(video_create_optional_params.get("seconds")) + resolution: Final = self._map_size_to_resolution(video_create_optional_params.get("size")) + input_reference: Final = video_create_optional_params.get("input_reference") + + translated: Final = { # mutable-ok: JSON request payload + key: value + for key, value in ( + ("duration", duration), + ("resolution", resolution), + ( + "keyframes", + [input_reference] # mutable-ok: JSON request payload + if input_reference is not None + else None, # mutable-ok: JSON request payload + ), + ) + if value is not None + } + passthrough: Final = { # mutable-ok: JSON request payload + key: value + for key, value in video_create_optional_params.items() + if key not in supported and key not in remapped + } + + return { # mutable-ok: JSON request payload + **translated, + **passthrough, + } # mutable-ok: JSON request payload + + def _map_duration(self, seconds: object) -> int | None: + if not isinstance(seconds, (int, float, str)): + return None + try: + duration: Final = 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, # mutable-ok: BaseVideoConfig signature + model: str, + api_key: str | None = None, + litellm_params: GenericLiteLLMParams | None = None, + ) -> dict: # mutable-ok: BaseVideoConfig signature + # Bind a new name rather than rebinding the caller's parameter. + request_api_key: Final = api_key or (litellm_params.api_key if litellm_params else None) + + # Resolve BFL credentials only. Falling back to ``litellm.api_key`` would + # send another provider's generic key to api.bfl.ai. + final_api_key: Final = ( + request_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( + { # mutable-ok: JSON request payload + "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, # mutable-ok: BaseVideoConfig signature + ) -> str: + base_url: Final[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: Final = 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())}" # mutable-ok: one-shot, for an error message + ) + + def transform_video_create_request( + self, + model: str, + prompt: str, + api_base: str, + video_create_optional_request_params: dict, # mutable-ok: BaseVideoConfig signature + litellm_params: GenericLiteLLMParams, + headers: dict, # mutable-ok: BaseVideoConfig signature + ) -> tuple[dict, list, str]: # mutable-ok: BaseVideoConfig signature + base_request: Final = { # mutable-ok: JSON request payload + "prompt": prompt, + **video_create_optional_request_params, + } # mutable-ok: JSON request payload + mode: Final = self._infer_mode(base_request) + + # draft_enhance re-renders a cached draft, so it carries no prompt. + request_data: Final = { # mutable-ok: JSON request payload + **{ # mutable-ok: JSON request payload + key: value for key, value in base_request.items() if not (mode == "draft_enhance" and key == "prompt") + }, # mutable-ok: JSON request payload + "mode": mode, + } + + url: Final = f"{api_base}{self._get_model_endpoint(model)}" + return request_data, [], url # mutable-ok: JSON request payload + + def _infer_mode(self, request_data: Mapping[str, Any]) -> 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, # mutable-ok: BaseVideoConfig signature + ) -> 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) + + # The region has to outlive this response, and the id is the only value + # the status and content calls receive. + regional_job_id: Final = _pack_region(job_id, polling_url) + + video_obj: Final = VideoObject( + id=regional_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"]) + + if custom_llm_provider: + video_obj.id = encode_video_id_with_provider(regional_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, # mutable-ok: BaseVideoConfig signature + ) -> tuple[str, dict]: # mutable-ok: BaseVideoConfig signature + return self._get_result_url(video_id, api_base), {} # mutable-ok: JSON request payload + + 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 = { # mutable-ok: JSON request payload + "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 = { # mutable-ok: VideoObject field + "credits": cost + } # mutable-ok: JSON request payload + + if custom_llm_provider and video_obj.id: + # Re-attach the region: BFL echoes a bare job id, but the caller may + # reuse this id for content retrieval and must land on the same host. + polled_url: Final = self._polled_url(raw_response) + video_obj.id = encode_video_id_with_provider( + _pack_region(video_obj.id, polled_url), custom_llm_provider, None + ) + + return video_obj + + def _polled_url(self, raw_response: httpx.Response) -> str | None: + """The URL this response came from, when httpx recorded one. + + ``Response.request`` raises rather than returning None when the response + was built without a request, which is the case in unit tests. + """ + try: + return str(raw_response.request.url) + except RuntimeError: + return None + + 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: Final = 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, # mutable-ok: BaseVideoConfig signature + variant: str | None = None, + ) -> tuple[str, dict]: # mutable-ok: BaseVideoConfig signature + return self._get_result_url(video_id, api_base), {} # mutable-ok: JSON request payload + + 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: Mapping[str, Any]) -> str: + status: Final = response_data.get("status", "Pending") + result: Final[Mapping[str, Any]] = response_data.get("result") or MappingProxyType({}) + 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: + """Build the polling URL, honouring the region the job was dispatched to. + + The global host answers 404 for a regional job, so a packed id sends the + request back to the host BFL named at submission. + """ + original_video_id: Final = extract_original_video_id(video_id) + job_id, region_host = _unpack_region(original_video_id) + host: Final = f"https://{region_host}" if region_host else api_base.rstrip("/") + result_url: Final = f"{host}/v1/get_result?id={job_id}" + assert_bfl_polling_url(result_url) + return result_url + + def _parse_json(self, raw_response: httpx.Response) -> dict: # mutable-ok: decoded JSON body + 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, # mutable-ok: BaseVideoConfig signature + extra_body: dict[str, Any] | None = None, # mutable-ok: BaseVideoConfig signature + ) -> tuple[str, dict]: # mutable-ok: BaseVideoConfig signature + 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, # mutable-ok: BaseVideoConfig signature + after: str | None = None, + limit: int | None = None, + order: str | None = None, + extra_query: dict[str, Any] | None = None, # mutable-ok: BaseVideoConfig signature + ) -> tuple[str, dict]: # mutable-ok: BaseVideoConfig signature + 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]: # mutable-ok: BaseVideoConfig signature + 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, # mutable-ok: BaseVideoConfig signature + ) -> tuple[str, dict]: # mutable-ok: BaseVideoConfig signature + 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, # mutable-ok: BaseVideoConfig signature + ) -> BlackForestLabsError: + return BlackForestLabsError(status_code=status_code, message=error_message) diff --git a/litellm/utils.py b/litellm/utils.py index c2770a1a26d..7076db2ff34 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -9037,6 +9037,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 diff --git a/model_prices_and_context_window.json b/model_prices_and_context_window.json index a6da2c1fb09..3a2ad133653 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -12160,6 +12160,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", diff --git a/tests/test_litellm/llms/black_forest_labs/videos/__init__.py b/tests/test_litellm/llms/black_forest_labs/videos/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/test_litellm/llms/black_forest_labs/videos/test_bfl_video_transformation.py b/tests/test_litellm/llms/black_forest_labs/videos/test_bfl_video_transformation.py new file mode 100644 index 00000000000..3c66bfec170 --- /dev/null +++ b/tests/test_litellm/llms/black_forest_labs/videos/test_bfl_video_transformation.py @@ -0,0 +1,360 @@ +""" +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_region(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 video.status == "queued" + assert video.seconds == "8" + assert video.size == "fhd" + + # The region survives inside the id, which is all the status call gets. + status_url, _ = self.config.transform_video_status_retrieve_request( + video_id=video.id, + api_base=API_BASE, + litellm_params=None, + headers={}, + ) + assert status_url == POLLING_URL + + def test_status_url_falls_back_to_the_api_base_without_a_region(self): + video = self.config.transform_video_create_response( + model="flux-3-video", + raw_response=_response({"id": JOB_ID, "polling_url": None}), + logging_obj=self.mock_logging_obj, + custom_llm_provider="black_forest_labs", + ) + + assert extract_original_video_id(video.id) == JOB_ID + + status_url, _ = self.config.transform_video_status_retrieve_request( + video_id=video.id, + api_base=API_BASE, + litellm_params=None, + headers={}, + ) + assert status_url == f"{API_BASE}/v1/get_result?id={JOB_ID}" + + def test_content_request_targets_the_same_region_as_the_status_call(self): + video = 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", + ) + + content_url, _ = self.config.transform_video_content_request( + video_id=video.id, + api_base=API_BASE, + litellm_params=None, + headers={}, + ) + assert content_url == POLLING_URL + + def test_a_packed_region_outside_bfl_is_rejected(self): + with pytest.raises(BlackForestLabsError, match="not within the bfl.ai domain"): + self.config.transform_video_status_retrieve_request( + video_id=f"{JOB_ID}@attacker.example.com", + api_base=API_BASE, + litellm_params=None, + headers={}, + ) + + 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")